(.*?)", txt, re.S)
return {"title": _clean(t.group(1)) if t else "", "content": body, "link": url}
def _img_ok(self, u):
if not u or not u.startswith("http"):
return False
if "/wp-content/uploads/" not in u and "/uploads/" not in u:
return False
if not _IMG_OK.search(urlparse(u).path):
return False
if _IMG_BAD.search(u):
return False
return True
def _imgs(self, content_html):
"""正文 HTML → 有序去重的原图列表"""
if not content_html:
return []
h = html_lib.unescape(content_html)
full_of = {}
for m in re.finditer(r'
]+href="([^"]+\.(?:jpe?g|png|webp|gif|avif))"', h, re.I):
u = m.group(1)
if self._img_ok(u):
full_of[_stem_key(u)] = u
out, seen = [], set()
for m in re.finditer(r"]+>", h, re.I):
tag = m.group(0)
u = ""
for k in ("data-src", "data-lazy-src", "data-original", "src"):
mm = re.search(k + r'=["\']([^"\']+)["\']', tag, re.I)
if mm and mm.group(1).strip():
u = html_lib.unescape(mm.group(1).strip())
break
if not u:
continue
if not u.startswith("http"):
u = urljoin(SITE + "/", u)
if not self._img_ok(u):
continue
full = full_of.get(_stem_key(u))
if full and not _has_size(full): # 有真原图就用原图
u = full
if u in seen:
continue
seen.add(u)
out.append(u)
return out
def _post_imgs(self, vid, fresh=None):
"""拿一篇作品的图集。vid 可以是 p:123 也可以是 u:https://.../slug/
fresh=True → 这次现拉(不吃任何缓存)。点开看图就走这条:
点哪一套就现拉哪一套,不认位置、不认共享列表、不吃缓存。"""
vid = str(vid or "")
if "|" in vid:
vid = vid.split("|")[0]
if fresh is None:
fresh = not getattr(self, "cache", 0)
ttl = 0 if fresh else CACHE_DETAIL
post = None
if vid.startswith("u:"):
url = vid[2:]
slug = urlparse(url).path.strip("/").split("/")[-1]
if slug:
post = self._post_by_slug(slug, ttl)
if not post:
post = self._post_from_html(url, ttl)
else:
pid = vid.split(":")[-1]
if pid.isdigit():
post = self._post_by_id(pid, ttl)
if not post:
return "", [], ""
t = post.get("title")
title = _clean(t.get("rendered") if isinstance(t, dict) else t)
raw = post.get("content")
body = raw.get("rendered") if isinstance(raw, dict) else (raw or "")
imgs = self._imgs(body)
link = _clean(post.get("link")) or ""
if not imgs and link:
alt = self._post_from_html(link, ttl)
if alt:
imgs = self._imgs(alt.get("content") or "")
if not title:
title = _clean(alt.get("title"))
return title, imgs, link
def detailContent(self, ids):
"""详情层 —— 现在这是主路(卡片不挂 action 了)。
卡片点进来就落到这里:把图集给出来,同时**页面一起来就把看图窗弹出来**
(约 0.45 秒),用户感觉就是「点一下直接看图」。
客户端会同时开详情页/播放页 —— 关窗时垫在下面的页面一起收掉,返回一次回列表。"""
vid = ""
try:
vid = str(ids[0] if isinstance(ids, (list, tuple)) else ids or "")
except Exception:
vid = ""
if not vid:
return {"list": []}
base = vid.split("|")[0]
parts = vid.split("|")
qkey = parts[1] if len(parts) > 1 else ""
try:
pos = int(parts[2]) if len(parts) > 2 else 0
except Exception:
pos = 0
title, imgs, link = self._post_imgs(base)
if not imgs:
return {"list": []}
cover = imgs[0]
if not self._q_get(qkey):
# 队列没了(客户端重启 / 从别的入口点进来)→ 现开一条只含这一篇的,
# 保证点了一定能弹看图窗
qkey = self._q_new([base], 0, title or "", None)
pos = 0
content = title or "XiuRen"
content += "\n图片:%d 张" % len(imgs)
if link:
content += "\n" + link
item = {
"vod_id": vid,
"vod_name": title or "XiuRen",
"vod_pic": cover,
"vod_remarks": "共 %d 张" % len(imgs),
"vod_actor": "",
"vod_content": content.strip(),
"pictures": imgs,
}
if self.line:
if self.webview:
# 旧版:走私有 WebView 桥(需要客户端支持 ZAKA_APPEND 等能力)
item["vod_play_from"] = "🖼 查看图片"
# 把作品 id 直接写进线路:点哪套就是哪套(队列只用来"往下接")
item["vod_play_url"] = "查看图片$img|%s|%s|%d" % (base, qkey, max(0, pos))
# 页面起来之后自己弹看图页(不用再点一次「查看图片」)
self._auto_open(vid, title)
else:
# 默认:按 base.spider 图文源契约输出 pics:// 播放串。
# 多张图以 && 分隔,主流 TVBox / 影视仓 都会以画廊/幻灯片形式展示,
# 不依赖任何私有 WebView 桥,兼容性最好。
play_url = "看图$pics://" + "&&".join(imgs)
item["vod_play_from"] = "🖼 看图"
item["vod_play_url"] = play_url
return {"list": [item], "parse": 0, "jx": 0}
# ==================== action:点卡片直接进看图页 ====================
def action(self, action, value):
a = str(action or "").strip().lower()
v = str(value or "").strip()
if a in ("img", "pic", "photo", "view", "看图"):
try:
self._open_from_card(v)
except Exception as e:
self._toast("看图页没弹出来:" + str(e)[:60])
return ""
# 兜底:万一客户端把 folder 卡片的 vod_id 当 action 派发过来
if v.startswith("c:") or v.startswith("l:") or v.startswith("t:"):
return ""
return ""
def _open_from_card(self, vid):
"""从卡片直接开看图页。动作全丢后台,别卡住客户端线程。"""
def work():
base = vid.split("|")[0]
parts = vid.split("|")
qkey = parts[1] if len(parts) > 1 else ""
try:
pos = int(parts[2])
except Exception:
pos = 0
q = self._q_get(qkey)
title = (q.get("title") if q else "") or "看图"
# 只把作品 id 交给看图页 —— 队列在不在都不影响取图
self._popup_gallery(title, base, qkey, pos)
try:
threading.Thread(target=work, daemon=True).start()
except Exception:
work()
def _auto_open(self, vid, title):
"""详情页兜底弹窗
默认(auto=2):等 2.5 秒 —— 这 2.5 秒里客户端基本已经靠「自动播图片线路」
把看图窗弹出来了(和 Pinterest 一模一样的路),那就什么都不做(不抢、不叠、不闪);
只有极少数不自动播线路的客户端才补弹一次,保证一定看得到图。
auto=1 回到上一版「详情页一起来 0.45 秒就弹」,auto=0 完全不补。
"""
if not self.auto:
return
def work():
time.sleep(0.45 if self.auto == 1 else 2.5)
base = vid.split("|")[0]
parts = vid.split("|")
qkey = parts[1] if len(parts) > 1 else ""
try:
pos = int(parts[2]) if len(parts) > 2 else 0
except Exception:
pos = 0
# 客户端已经靠「自动播图片线路」把窗弹出来过了 → 不重复弹(不闪)
try:
with self._pop_lock:
if self._pop_alive and self._pop_key == base:
return
except Exception:
pass
ok = False
try:
ok = self._popup_gallery(title or "", base, qkey, pos,
chain_close=True, quick=False)
except Exception:
ok = False
if not ok:
# 窗没真弹出来(页面还没就绪 / 被盖住)→ 隔一秒再怼一次,不静默失败
try:
time.sleep(1.0)
self._popup_gallery(title or "", base, qkey, pos,
chain_close=True, quick=False)
except Exception:
pass
try:
threading.Thread(target=work, daemon=True).start()
except Exception:
pass
# ==================== 看图页(原生弹窗 + WebView) ====================
def _popup_gallery(self, title, postkey, qkey="", pos=0, chain_close=False, quick=True):
"""开看图窗(第一屏只认作品 id)。
v3.0 病根:第一屏以前是去**共享队列**里拿"当前攒下来的那批图"——
队列是整列作品共用的,攒出来的又永远从第 0 张开始 →
在同一个列表里点不同的几套,弹出来全是同一套;队列一重建又集体换一套
(于是"过一阵子再打开变成别的内容")。
现在:弹窗前拿**这一套的作品 id** 现拉这一套的图;
队列/位置只用来决定「往下接哪一套」,再也不参与取图。
窗还开着又点了另一套 → 原地换内容(ZAKA_RESET),不叠第二层窗。"""
postkey = str(postkey or "")
qkey = str(qkey or "")
if not postkey and qkey:
# 兼容手里还捏着旧卡片的:只有「队列 + 位置」,按位置去队列里查出作品 id
q0 = self._q_get(qkey)
items = list((q0.get("items") if q0 else None) or [])
if items:
try:
k0 = max(0, min(int(pos or 0), len(items) - 1))
except Exception:
k0 = 0
postkey = str(items[k0])
if not postkey:
return False
# 1) 现拉这一套(认作品 id —— 不认位置、不认共享列表、不吃缓存)
name, imgs = "", []
try:
name, imgs, _link = self._post_imgs(postkey, fresh=True)
except Exception:
name, imgs = "", []
name = _clean(name) or _clean(title) or "看图"
if not imgs:
return False
# 2) 位置只用来决定「往下接哪一套」:从这一套的下一条开始
self._vm = None
q = self._q_get(qkey) if qkey else None
if q:
try:
k = list(q.get("items") or []).index(postkey)
except Exception:
k = -1
if k >= 0:
q["i"] = k + 1
q["photos"] = []
self._vm = q
recs = [{"u": u, "c": self._chain(u), "w": name[:40]} for u in imgs]
btitle = "%s · 本套 %d 张" % (name, len(recs))
key = postkey
now = time.time()
with self._pop_lock:
if self._pop_alive and self._pop_key == key:
return True # 窗开着,看的就是这一套
swap = self._pop_js if (self._pop_alive and self._pop_js) else None
if swap is None:
if now - self._pop_busy < 4:
return False
self._pop_busy = now
if swap is not None:
# 窗开着但看的是另一套 → 原地把内容换掉(不叠窗、不闪、不重开)
try:
swap("window.ZAKA_RESET&&window.ZAKA_RESET(%s,%s)"
% (json.dumps(recs, ensure_ascii=False),
json.dumps(btitle, ensure_ascii=False)))
with self._pop_lock:
self._pop_key = key
return True
except Exception:
pass
html = self._gallery_html(btitle, recs)
try:
return self._popup(html, key, chain_close=chain_close, title=btitle,
quick=quick, pre=True)
finally:
with self._pop_lock:
self._pop_busy = 0.0
def _esc(self, s):
"""HTML 转义(看图页标题要用)"""
s = "" if s is None else str(s)
try:
return html_lib.escape(s)
except Exception:
return (s.replace("&", "&").replace("<", "<")
.replace(">", ">").replace('"', """))
def _slide_rec(self, rec):
"""一张图 → 看图页要的 {u: 主链, c: 备用链(| 分隔), w: 落款}"""
return {"u": str(rec.get("u") or ""),
"c": str(rec.get("c") or ""),
"w": str(rec.get("w") or "")}
# ==================== 内置黑屏占位视频(治「正在切换站源」) ====================
def _ph_dir(self):
"""放小文件的目录:优先应用缓存目录(可写就缓存住),拿不到再退临时目录
临时目录**不缓存** —— 安卓上它常常是不可写的 /tmp,万一先走到这条路,
缓存住就再也升不回应用缓存目录了(占位片写不出来 → 又回 about:blank → 又切源)。"""
if _PH_DIR_CACHE[0]:
return _PH_DIR_CACHE[0]
try:
act = self._activity()
if act is not None:
d = act.getCacheDir().getAbsolutePath()
if d:
_PH_DIR_CACHE[0] = str(d)
return str(d)
except Exception:
pass
try:
import tempfile
d = tempfile.gettempdir()
try: # 探一下真能写才用
probe = os.path.join(d, ".zaka_w")
with open(probe, "wb") as f:
f.write(b"1")
os.remove(probe)
return d
except Exception:
return ""
except Exception:
return ""
def _ph_http(self, path):
"""本机回环小服务:把占位片用 http://127.0.0.1:端口/xxx.mp4 发出去(起不来返回空串)
为什么不用 file://:个别播放器的数据源只认 http(s),file:// 播不动 → 又去切站源。
回环 http 谁都认;片子 8KB,一次读完,后面 4 小时不再要数据,几乎不吃流量。"""
with _PH_LOCK:
if _PH_SRV["url"]:
return _PH_SRV["url"]
try:
import http.server
try:
with open(path, "rb") as f:
body = f.read()
except Exception:
return ""
total = len(body)
class Handler(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *a):
pass
def _head(self, code, start, end):
self.send_response(code)
self.send_header("Content-Type", "video/mp4")
self.send_header("Accept-Ranges", "bytes")
self.send_header("Content-Length", str(max(0, end - start + 1)))
if code == 206:
self.send_header("Content-Range",
"bytes %d-%d/%d" % (start, end, total))
self.end_headers()
def _rng(self):
r = str(self.headers.get("Range") or "").strip().lower()
if not r.startswith("bytes=") or total <= 0:
return None
piece = r[6:].split(",")[0].strip()
a, _sep, b = piece.partition("-")
try:
if a == "":
s2 = max(0, total - int(b))
e2 = total - 1
else:
s2 = int(a)
e2 = int(b) if b else total - 1
except Exception:
return None
s2 = max(0, min(s2, total - 1))
e2 = max(s2, min(e2, total - 1))
return s2, e2
def do_GET(self):
try:
r = self._rng()
if r:
self._head(206, r[0], r[1])
self.wfile.write(body[r[0]:r[1] + 1])
return
self._head(200, 0, total - 1)
self.wfile.write(body)
except Exception:
pass
def do_HEAD(self):
try:
self._head(200, 0, total - 1)
except Exception:
pass
srv = None
for nm in ("ThreadingHTTPServer", "HTTPServer"):
cls = getattr(http.server, nm, None)
if cls is None:
continue
try:
srv = cls(("127.0.0.1", 0), Handler)
break
except Exception:
srv = None
if srv is None:
return ""
port = int(srv.server_address[1])
th = threading.Thread(target=srv.serve_forever)
th.daemon = True
th.start()
url = "http://127.0.0.1:%d/%s" % (port, PH_NAME)
with _PH_LOCK:
_PH_SRV["srv"] = srv
_PH_SRV["url"] = url
return url
except Exception:
return ""
def _ph_video(self):
"""图片线路回一条**真能播**的视频地址(黑屏 / 无声 / 4 小时 / 8KB)
病根:详情页第一条线路就是「🖼 查看图片」,客户端进场自动播它。
以前回 about:blank 或图片直链 → 播放器一播就失败 → 客户端判定"坏源" →
自己去找别的站源(「正在切换站源」一直闪)。给它一条播得动的,播放器不报错,
自然就不会切源。优先本机 http(播放器最认),退一步 file://。"""
if not self.ph:
return ""
try:
with self._ph_lock:
d = self._ph_dir()
if not d:
return ""
p = os.path.join(d, PH_NAME)
ok = False
try:
ok = os.path.exists(p) and os.path.getsize(p) == PH_SIZE
except Exception:
ok = False
if not ok:
with open(p, "wb") as f:
f.write(base64.b64decode(PH_MP4_B64))
url = self._ph_http(p)
if url:
return url
return "file://" + p
except Exception:
return ""
# ==================== 看图模式记忆 ====================
def _mode_file(self):
d = self._ph_dir()
return os.path.join(d, "zaka_view.txt") if d else ""
def _load_mode(self):
"""上次在看图页里选的那个模式,重启也记得"""
try:
p = self._mode_file()
if p and os.path.exists(p):
with open(p) as f:
self.view_mode = _int(f.read().strip(), self.view_mode, 0, 2)
except Exception:
pass
def _save_mode(self):
try:
p = self._mode_file()
if p:
with open(p, "w") as f:
f.write(str(int(self.view_mode or 0)))
except Exception:
pass
def _gallery_html(self, title, slides=None):
"""首屏图片直接塞进页面(JSON)—— 弹窗一出来就有图,不指望客户端回调。
v2.2:三种看图模式(左右 / 上下无缝 / 瀑布流)都在页面里,右上角点一下切。"""
slides = list(slides or [])
data = [self._slide_rec(r) for r in slides]
try:
blob = json.dumps(data, ensure_ascii=False).replace("= 2:
act = mid[0][1]
break
if live and i >= 4:
# 没等到"像详情页"的,也别让用户干等 —— 挂当前这一页
act = live[0][1]
break
time.sleep(0.15)
if act is None:
acts = self._all_activities()
if acts:
for c, a, _p in acts:
if any(k in c for k in plyk):
act = a
break
else:
act = ([x for x in acts if not x[2]] or acts)[0][1]
if act is None:
return False
return self._popup_ui(act, html, qkey, chain_close, title, pre)
def _popup_ui(self, act, html, qkey, chain_close=False, title="", pre=False):
try:
from java import jclass, dynamic_proxy
from java.lang import Runnable
spider = self
self._popup_host = act # 记住这一页:关弹窗时先收它(不依赖客户端类名)
# pre=True:首屏图已经写进页面了,别再白拉一次接口
state = {"qkey": qkey, "loaded": [bool(pre)], "shown": False}
class Run(dynamic_proxy(Runnable)):
def __init__(self, fn):
super().__init__()
self.fn = fn
def run(self):
self.fn()
def ui():
Dialog = jclass("android.app.Dialog")
WebView = jclass("android.webkit.WebView")
FrameLayout = jclass("android.widget.FrameLayout")
Button = jclass("android.widget.Button")
ColorDrawable = jclass("android.graphics.drawable.ColorDrawable")
Color = jclass("android.graphics.Color")
LP = jclass("android.widget.FrameLayout$LayoutParams")
Gravity = jclass("android.view.Gravity")
LinearLayout = jclass("android.widget.LinearLayout")
LLP = jclass("android.widget.LinearLayout$LayoutParams")
d = Dialog(act)
d.requestWindowFeature(1)
w = WebView(act)
ws = w.getSettings()
ws.setJavaScriptEnabled(True)
ws.setDomStorageEnabled(True)
ws.setUserAgentString(UA)
w.setBackgroundColor(Color.BLACK)
try:
ws.setLoadWithOverviewMode(True)
ws.setMediaPlaybackRequiresUserGesture(False)
except Exception:
pass
def js(code):
def go():
try:
w.evaluateJavascript(code, None)
except Exception:
pass
dispatch(go)
def js_async(code):
def go():
try:
w.evaluateJavascript(code, None)
except Exception:
pass
try:
act.runOnUiThread(Run(go))
except Exception:
try:
act.getWindow().getDecorView().post(Run(go))
except Exception:
pass
def dispatch(fn):
try:
act.runOnUiThread(Run(fn))
except Exception:
try:
act.getWindow().getDecorView().post(Run(fn))
except Exception:
pass
def push(data, msg=""):
"""把一批图推给页面(空的话就说一句)"""
if data:
js_async("window.ZAKA_APPEND&&window.ZAKA_APPEND(%s)"
% json.dumps(data, ensure_ascii=False))
elif msg:
js_async("window.ZAKA_END&&window.ZAKA_END(%s)"
% json.dumps(str(msg), ensure_ascii=False))
gate = {"lock": threading.Lock(), "last": 0.0}
def load_more(note=True):
"""加载下一组作品,推给页面"""
with gate["lock"]:
now = time.time()
if now - gate["last"] < 0.8:
return
gate["last"] = now
def work():
try:
data, end = spider._vm_next()
except Exception:
data, end = [], False
if data:
push(data)
elif end:
push(None, "已经到底了")
else:
push(None, "")
js_async("window.ZAKA_INFO&&window.ZAKA_INFO('这一组没拉到图,滑一下再试')")
try:
threading.Thread(target=work, daemon=True).start()
except Exception:
work()
def first_load():
with gate["lock"]:
if state["loaded"][0]:
return
state["loaded"][0] = True
load_more()
# 页面加载完 → 拉第一组;同时用定时器兜底(有的客户端 onPageFinished 不响)
try:
WC = jclass("android.webkit.WebViewClient")
class Cli(dynamic_proxy(WC)):
def __init__(self):
super().__init__()
def onPageFinished(self, view, url):
try:
first_load()
except Exception:
pass
w.setWebViewClient(Cli())
except Exception:
pass
try:
threading.Timer(0.45, first_load).start()
except Exception:
pass
# 页面滑到底 → console.log('ZAKA_MORE') → 这里接住去加载下一组
try:
CC = jclass("android.webkit.WebChromeClient")
class Chrome(dynamic_proxy(CC)):
def __init__(self):
super().__init__()
def onReceivedTitle(self, view, t):
try:
s = str(t or "")
if s.startswith("IDX"):
spider._cur_idx = int((s[3:].strip() or "0"))
except Exception:
pass
def onConsoleMessage(self, cm):
try:
msg = str(cm.message() or "")
except Exception:
return True
if "ZAKA_MORE" in msg:
try:
load_more()
except Exception:
pass
if "ZAKA_CUR:" in msg:
try:
spider._cur_url = msg.split("ZAKA_CUR:", 1)[1].strip()
except Exception:
pass
if "ZAKA_SETMODE:" in msg:
try:
spider.view_mode = int(
msg.split("ZAKA_SETMODE:", 1)[1].strip()[0])
spider._save_mode()
except Exception:
pass
return True
w.setWebChromeClient(Chrome())
except Exception:
pass
frame = FrameLayout(act)
try:
frame.addView(w, LP(-1, -1))
except Exception:
pass
try:
w.loadDataWithBaseURL(SITE + "/", html, "text/html", "utf-8", None)
except Exception:
pass
def cur_photo():
# 页面上正显示哪张,页面自己会回报 —— 不再靠"第几个位置"去猜
u0 = str(getattr(spider, "_cur_url", "") or "")
if u0:
return {"u": u0}
q = spider._q_get(state["qkey"]) or {}
ph = q.get("photos") or []
try:
i = int(getattr(spider, "_cur_idx", 0))
except Exception:
i = 0
if not (0 <= i < len(ph)):
i = 0
return ph[i] if ph else {}
def do_dl():
cur = cur_photo()
u = str(cur.get("u") or "")
if not u:
spider._toast("还没加载出图片")
return
def work():
try:
ok, m = spider._download(u)
spider._toast(("✅ " if ok else "❌ ") + m)
except Exception as e:
spider._toast("下载没起来:" + str(e)[:40])
try:
threading.Thread(target=work, daemon=True).start()
except Exception:
work()
class Click(dynamic_proxy(jclass("android.view.View$OnClickListener"))):
def __init__(self, fn):
super().__init__()
self.fn = fn
def onClick(self, v):
try:
self.fn()
except Exception:
pass
d.setContentView(frame)
# 关窗回调必须**总是**挂:① 记下「窗已经关了」(防叠窗判据靠它)
# ② 需要时把垫在下面的详情页一起收掉 —— 返回一次就回列表
try:
DOL = jclass("android.content.DialogInterface$OnDismissListener")
class Dismiss(dynamic_proxy(DOL)):
def __init__(self):
super().__init__()
def onDismiss(self, dlg):
try:
spider._pop_alive = False
spider._pop_js = None
spider._pop_key = ""
except Exception:
pass
if chain_close:
try:
spider._close_middle()
except Exception:
pass
d.setOnDismissListener(Dismiss())
except Exception:
pass
d.show()
try:
state["shown"] = True # v2.1:窗真出来了,拿它判断要不要重试
except Exception:
pass
# 窗真出来了才算「开着」—— 没弹出来不算,别把下一次点击吞掉
try:
spider._pop_alive = True
spider._pop_key = qkey
spider._popup_host = act
spider._pop_js = js_async # 换另一套时原地换内容,不叠窗
except Exception:
pass
# 按钮栏是「装饰」——放到弹窗出来之后再搭,它出问题也不影响看图
try:
dens = 1.0
try:
dens = float(act.getResources().getDisplayMetrics().density)
except Exception:
pass
bar = LinearLayout(act)
bar.setOrientation(1)
def mk_btn(text, fn, first=False):
b = Button(act)
b.setText(text)
try:
b.setTextSize(12)
b.setTextColor(Color.WHITE)
b.setBackgroundColor(Color.argb(215, 26, 26, 26))
except Exception:
pass
try:
b.setOnClickListener(Click(fn))
except Exception:
pass
bp = LLP(-2, -2)
try:
if not first:
bp.setMargins(0, int(8 * dens), 0, 0)
except Exception:
pass
bar.addView(b, bp)
return b
# v2.5:按钮直接写着当前是哪种模式,点一下换下一种(盒子遥控器也能按)
mnames = ["\u2194 \u5de6\u53f3", "\u2195 \u4e0a\u4e0b", "\u25a6 \u6c34\u5e03"]
mstate = [int(spider.view_mode or 0) % 3]
def next_mode():
mstate[0] = (mstate[0] + 1) % 3
try:
mbtn.setText("\U0001F500 \u770b\u56fe: " + mnames[mstate[0]])
except Exception:
pass
js_async("window.ZAKA_MODE&&window.ZAKA_MODE()")
mbtn = mk_btn("\U0001F500 \u770b\u56fe: " + mnames[mstate[0]], next_mode, True)
mk_btn("⬇️ 下载", do_dl)
mk_btn("⬇️➡️ 继续加载", lambda: load_more())
lp = LP(-2, -2, Gravity.BOTTOM | Gravity.RIGHT)
try:
lp.setMargins(0, 0, int(16 * dens), int(74 * dens))
except Exception:
pass
frame.addView(bar, lp)
except Exception:
pass
try:
win = d.getWindow()
if win:
win.getDecorView().setPadding(0, 0, 0, 0)
win.setBackgroundDrawable(ColorDrawable(Color.TRANSPARENT))
win.setLayout(-1, -1)
except Exception:
pass
act.getWindow().getDecorView().post(Run(ui))
# 等 UI 线程把窗搭出来(最多 1.2 秒):真出来了才算成功,失败好重试
for _ in range(12):
if state.get("shown"):
return True
time.sleep(0.1)
return bool(state.get("shown"))
except Exception as e:
self._toast("看图页打不开:" + str(e)[:40])
return False
def _close_middle(self, max_depth=3):
"""把垫在弹窗下面的详情页 / 播放页收掉(主列表绝不动)"""
KEYS = ("detail", "player", "playback", "vod", "episode", "album")
def is_home(act):
try:
pm = act.getPackageManager()
it = pm.getLaunchIntentForPackage(str(act.getPackageName()))
home = str(it.getComponent().getClassName())
return str(act.getClass().getName()) == home
except Exception:
return False
def step(n, first=False):
if n <= 0:
return
act = self._activity()
if act is None:
try:
threading.Timer(0.35, step, args=(n - 1, first)).start()
except Exception:
pass
return
if first:
host = getattr(self, "_popup_host", None)
if host is not None:
self._popup_host = None
try:
if not is_home(host):
host.finish()
try:
threading.Timer(0.35, step, args=(n - 1, False)).start()
except Exception:
pass
return
except Exception:
pass
try:
cls = str(act.getClass().getName()).lower()
except Exception:
return
if is_home(act) or not any(k in cls for k in KEYS):
return
try:
act.finish()
except Exception:
return
try:
threading.Timer(0.35, step, args=(n - 1, False)).start()
except Exception:
pass
try:
threading.Timer(0.12, step, args=(max_depth, True)).start()
except Exception:
pass
def _all_activities(self):
"""当前活着的 Activity 列表:[(类名(小写), activity, 是否 paused)]"""
out = []
try:
from java import jclass
JClass = jclass("java.lang.Class")
AT = JClass.forName("android.app.ActivityThread")
cur = AT.getMethod("currentActivityThread").invoke(None)
f = AT.getDeclaredField("mActivities")
f.setAccessible(True)
for r in f.get(cur).values().toArray():
rc = r.getClass()
pf = rc.getDeclaredField("paused")
pf.setAccessible(True)
af = rc.getDeclaredField("activity")
af.setAccessible(True)
a = af.get(r)
if a is None:
continue
try:
if a.isFinishing():
continue
except Exception:
pass
try:
cls = str(a.getClass().getName()).lower()
except Exception:
cls = ""
out.append((cls, a, bool(pf.getBoolean(r))))
except Exception:
pass
return out
def _activity(self, prefer=None):
acts = self._all_activities()
if not acts:
return None
live = [x for x in acts if not x[2]]
pool = live or acts
if prefer:
for c, a, _p in pool:
if any(k in c for k in prefer):
return a
return pool[0][1]
def _toast(self, msg):
try:
from java import jclass, dynamic_proxy
from java.lang import Runnable
act = self._activity()
if not act:
return
Toast = jclass("android.widget.Toast")
class Run(dynamic_proxy(Runnable)):
def __init__(self, fn):
super().__init__()
self.fn = fn
def run(self):
self.fn()
act.getWindow().getDecorView().post(Run(lambda: Toast.makeText(act, str(msg), 1).show()))
except Exception:
pass
def _download(self, url, name=""):
"""存图:走系统下载器,图片落 Pictures/XiuRen"""
try:
from java import jclass
Context = jclass("android.content.Context")
Request = jclass("android.app.DownloadManager$Request")
Uri = jclass("android.net.Uri")
Environment = jclass("android.os.Environment")
act = self._activity()
if act is None:
return False, "没找到可用的页面"
dm = act.getSystemService(Context.DOWNLOAD_SERVICE)
fn = name or (urlparse(url).path.rsplit("/", 1)[-1] or ("xiuren_%d.jpg" % int(time.time())))
req = Request(Uri.parse(url))
req.setTitle(fn)
req.setNotificationVisibility(1)
try:
req.setDestinationInExternalPublicDir(Environment.DIRECTORY_PICTURES, "XiuRen/" + fn)
except Exception:
pass
try:
req.addRequestHeader("User-Agent", UA)
req.addRequestHeader("Referer", SITE + "/")
except Exception:
pass
try:
act.runOnUiThread(_JRun(lambda: dm.enqueue(req)))
except Exception:
dm.enqueue(req)
return True, "已开始下载:" + fn
except Exception as e:
return False, "下载没起来:" + str(e)[:40]
# ==================== 播放(图片直链) ====================
def playerContent(self, flag, id, vipFlags=None):
url = str(id or "")
# 有些客户端把详情页整行(「查看图片$img|xxx|0」)原样塞进来 → 只取 $ 后面那段
if "$" in url and ("img|" in url or url.startswith("查看图片")):
url = url.split("$")[-1]
header = {"User-Agent": UA, "Referer": SITE + "/"}
blank = {"parse": 0, "playUrl": "", "url": "about:blank", "header": header}
# ---- 图文源 pics:// 播放串:直接透传,由播放器以画廊展示 ----
# 兼容某些把整串(来源名$pics://...)当 id 传进来的客户端
if "pics://" in url:
p = url.find("pics://")
return {"parse": 0, "playUrl": "", "url": url[p:], "header": header}
# ---- 🖼 看图线路:点「查看图片」当场弹原生看图窗(和 Pinterest2 同一条路)----
if url.startswith("img|"):
parts = url.split("|")
a1 = parts[1] if len(parts) > 1 else ""
post, qkey, pos = "", "", 0
if a1.startswith("p:") or a1.startswith("u:"):
# 新格式:img|<作品id>|<队列key>|<位置> —— 直接认作品 id
post = a1
qkey = parts[2] if len(parts) > 2 else ""
try:
pos = int(parts[3]) if len(parts) > 3 else 0
except Exception:
pos = 0
else:
# 老格式:img|<队列key>|<位置>[|<作品id>] —— 手里还是旧卡片的也能看
qkey = a1
try:
pos = int(parts[2]) if len(parts) > 2 else 0
except Exception:
pos = 0
if len(parts) > 3:
post = parts[3]
if not post and not self._q_get(qkey):
return blank # 两手空空就别弹了,详情层还会补一次
self._popup_gallery("", post, qkey, pos, chain_close=True, quick=False)
# v2.5:默认回**本机黑屏占位片**(真能播 -> 播放器不报错 -> 不切源);
# ph=0 才回 Pinterest 那种 about:blank(拿不到本机 http 时自动降级到它)
ph = self._ph_video()
if ph:
return {"parse": 0, "playUrl": "", "url": ph, "header": header}
return blank
if url.startswith("u:") or url.startswith("p:"):
_t, pics, _l = self._post_imgs(url)
url = pics[0] if pics else ""
return {
"parse": 0,
"playUrl": "",
"url": url,
"header": header,
}
def isVideoFormat(self, url):
u = str(url or "").lower()
return any(x in u for x in (".mp4", ".m3u8", ".mkv", ".flv", ".avi", ".ts"))
def manualVideoCheck(self):
return False
def destroy(self):
with self._lock:
self._cache.clear()
self._queues.clear()
try:
if self._sess is not None:
self._sess.close()
except Exception:
pass
def _JRun(fn):
"""给 _download 用的小包装(拿不到 java 环境时直接调用)"""
try:
from java import dynamic_proxy
from java.lang import Runnable
class R(dynamic_proxy(Runnable)):
def __init__(self):
super().__init__()
def run(self):
try:
fn()
except Exception:
pass
return R()
except Exception:
return fn
# ==================== 本地自测(独立跑用,客户端不执行这段) ====================
if __name__ == "__main__":
import sys
sp = Spider()
sp.init(sys.argv[1] if len(sys.argv) > 1 else "")
hc = sp.homeContent()
print("[home] 分类数 = %d" % len(hc["class"]))
for c in hc["class"][:6]:
print(" ", c["type_id"], c["type_name"])
for tid in ("latest", "hot"):
r = sp.categoryContent(tid, 1, False, None)
print("[cat %-6s] %d 条 / pagecount=%s action=%s" %
(tid, len(r["list"]), r["pagecount"],
(r["list"][0].get("action") if r["list"] else "-")))
for it in r["list"][:2]:
print(" -", it["vod_name"][:50], "|", (it["vod_pic"] or "")[:70])
print(" id:", it["vod_id"])
big = hc["class"][2]["type_id"]
t0 = time.time()
r = sp.categoryContent(big, 1, False, None)
print("[大分类 %s] 小分类 %d 个, %.2fs" % (big, len(r["list"]), time.time() - t0))
for it in r["list"][:8]:
print(" -", it["vod_name"][:30], "|", it.get("vod_remarks"), "|", it["vod_id"])
wall = {"list": []}
if len(r["list"]) > 1:
sub = r["list"][2]["vod_id"]
t0 = time.time()
wall = sp.categoryContent(sub, 1, False, None)
print("[小分类 %s] 图片墙 %d 条, %.2fs" % (sub, len(wall["list"]), time.time() - t0))
if wall["list"]:
print(" 首条:", wall["list"][0]["vod_name"][:46], "| id:", wall["list"][0]["vod_id"])
# 看图队列:从当前这列连拉两组,验证"滑到底自动接下一组"
if wall["list"]:
v = wall["list"][0]["vod_id"]
parts = v.split("|")
q = sp._q_get(parts[1])
sp._vm = q
t0 = time.time()
a, e = sp._vm_next()
b, e2 = sp._vm_next()
print("[看图队列] 第一组 %d 张 / 第二组 %d 张, %.2fs, 到底=%s"
% (len(a), len(b), time.time() - t0, e2))
if a:
print(" 首图:", a[0]["u"][:100])
print(" 标题:", a[0]["w"])
r = sp.searchContent("杨晨晨", False, "1")
print("[search] %d 条" % len(r["list"]))
tv, imgs, link = sp._post_imgs(v if wall["list"] else "p:0")
print("[图集] %s | %d 张 | %s" % (tv[:40], len(imgs), link[:60]))