# -*- coding: utf-8 -*-
import os
import re
import json
import time
import hashlib
import threading
import base64
import sys
import socket
import urllib.request
import urllib.error
import urllib.parse
import requests
try:
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
except ImportError:
from http.server import HTTPServer as ThreadingHTTPServer, BaseHTTPRequestHandler
from base.spider import Spider as BaseSpider
_API_B64 = "aHR0cHM6Ly9memw0ay54eXovZ2o="
DATA_API = base64.b64decode(_API_B64).decode()
ALLOW_DOMAIN = ""
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/199.0.0.0 Safari/537.3 6"
PUSH_FLAG = "__TB_PUSH__::"
PLAY_FLAG = "__TB_PLAY__::"
SHARE_FLAG = "__TB_SHARE__::"
LAST_CLASS_NAMES = {"其它", "其他", "其它分类", "其他分类"}
BROWSER_ONLY_NAMES = {"py生直链"}
BROWSER_ONLY_URL_KEYS = ("仓库.php",)
PY_OPS_PIC = "https://files.seeusercontent.com/2026/10/13/g0dS/1791894305322_edit_2712206973539.png"
PY_SOURCE_PIC = "https://files.seeusercontent.com/2026/10/15/Y7lq/p3-flow-imagex-signbyteimgcom_29.png"
GUIDE_IMAGE = "https://s2.loli.net/2025/05/27/WqbTuL3QcxZfRmN.jpg"
UPDATE_API = "https://raw.gitcode.com/u013624795/jk/raw/main/gx.json"
UPDATE_ENABLED = True
UPDATE_INTERVAL = 600
SHARE_PORT_START = 8099
SHARE_MAX_TRIES = 20
FILE_CHUNK = 64 * 1024
def _detect_storage_root():
candidates = []
external = str(os.environ.get("EXTERNAL_STORAGE", "")).strip()
if external:
candidates.append(external)
candidates.extend(("/sdcard", "/storage/emulated/0", os.path.expanduser("~/storage/shared")))
seen = set()
for candidate in candidates:
path = os.path.abspath(os.path.expanduser(candidate))
real = os.path.realpath(path)
if real in seen:
continue
seen.add(real)
if os.path.isdir(path):
return real
return os.path.abspath(external or "/sdcard")
def _detect_local_base(storage_root):
for name in ("tvbox", "TVBox"):
path = os.path.realpath(os.path.join(storage_root, name))
if os.path.isdir(path):
return path
return os.path.realpath(os.path.join(storage_root, "tvbox"))
STORAGE_ROOT = _detect_storage_root()
LOCAL_BASE = _detect_local_base(STORAGE_ROOT)
DEFAULT_PY_DIR = os.path.join(LOCAL_BASE, "py")
TV_DIR = os.path.join(STORAGE_ROOT, "TV")
FENGLING_PATH = os.path.join(TV_DIR, "丰之玲.json")
SINGLE_PATH = os.path.join(TV_DIR, ".single_config.json")
SETTINGS_PATH = os.path.join(TV_DIR, "py_scan.settings.json")
CHECK_RESULTS_PATH = os.path.join(TV_DIR, "py_check_results.json")
RECV_FOLDER = os.path.join(STORAGE_ROOT, "Download", "PY助手")
class Spider(BaseSpider):
VERSION = "v2.3.3"
TID_PY_OPS = "__py_ops__"
TID_PY_LIST = "__py_list__"
RESCAN_ID = "__py_scan_rescan__"
EDIT_DIR_ID = "__py_scan_edit_dir__"
CHECK_UPDATE_ID = "__py_check_update__"
CHECK_ALL_ID = "__py_scan_check_all__"
VIEW_REPORT_ID = "__py_view_report__"
ACTION_RESCAN = "py_scan_rescan"
ACTION_EDIT_DIR = "py_scan_edit_dir"
ACTION_CHECK_UPDATE = "py_check_update"
ACTION_PUSH_ONE_PREFIX = "py_push:"
ACTION_WEB_OPEN_PREFIX = "web_open:"
ACTION_BROWSER_OPEN_PREFIX = "browser_open:"
ACTION_CHECK_ALL = "py_scan_check_all"
ACTION_VIEW_REPORT = "py_scan_view_report"
ACTION_CHECK_ONE_PREFIX = "py_check:"
ACTION_SEND = "py_send_file"
ACTION_RECV = "py_recv_file"
MAX_SCAN_DEPTH = 8
MAX_SCAN_FILES = 3000
MAX_SOURCE_SIZE = 5 * 1024 * 1024
CHECK_TIMEOUT = 5.0
def getName(self):
return "Py助手" + self.VERSION
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def init(self, extend=""):
if getattr(self, "_inited", False):
return
self._inited = True
self.session = requests.Session()
self.session.headers.update({"User-Agent": UA})
self._cache = None
self._cur_url = ""
self._cur_name = ""
self.lock = threading.RLock()
self.scan_dir = DEFAULT_PY_DIR
self.py_sources = []
self._py_status = self._empty_py_status()
self._dialog_refs = []
self._last_update_check = 0.0
self._guide_bytes = None
self._guide_lock = threading.Lock()
self.check_results = {}
self.check_state = {
"status": "idle", "dialog": None, "results": [],
"total": 0, "done": 0, "current": "",
}
self._send_server = None
self._send_port = 0
self._send_file_path = ""
self._send_file_name = ""
self._last_recv_addr = ""
self._load_settings()
self._load_check_results()
try:
self._refresh()
except Exception as exc:
self._py_status["warnings"].append("启动扫描失败: {}".format(exc))
try:
self._check_update()
except Exception:
pass
try:
threading.Thread(target=self._preload_guide_image).start()
except Exception:
pass
def destroy(self):
try:
self._stop_send_server()
except Exception:
pass
return "destroy"
def _ensure_inited(self):
if not getattr(self, "_inited", False):
self.init("")
def _preload_guide_image(self):
try:
with self._guide_lock:
if self._guide_bytes:
return
req = urllib.request.Request(GUIDE_IMAGE, headers={"User-Agent": UA})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=8) as r:
data = r.read()
with self._guide_lock:
self._guide_bytes = data
except Exception:
pass
def _get_guide_bytes(self):
with self._guide_lock:
if self._guide_bytes:
return self._guide_bytes
try:
req = urllib.request.Request(GUIDE_IMAGE, headers={"User-Agent": UA})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=8) as r:
data = r.read()
with self._guide_lock:
self._guide_bytes = data
return data
except Exception:
return None
def _show_toast(self, msg):
try:
from java import dynamic_proxy, jclass
from java.lang import Runnable
except Exception:
return
try:
activity = self._current_android_activity(jclass)
except Exception:
return
class UIR(dynamic_proxy(Runnable)):
def run(self):
try:
Toast = jclass("android.widget.Toast")
Toast.makeText(activity, msg, Toast.LENGTH_LONG).show()
except Exception:
pass
activity.runOnUiThread(UIR())
def _parse_version(self, v):
v = str(v or "").strip().lstrip("vV")
parts = re.split(r"[.\-_]", v)
nums = []
for p in parts:
try:
nums.append(int(p))
except Exception:
m = re.match(r"\d+", p)
nums.append(int(m.group()) if m else 0)
return nums or [0]
def _is_newer(self, remote, local):
r = self._parse_version(remote)
l = self._parse_version(local)
n = max(len(r), len(l))
r = r + [0] * (n - len(r))
l = l + [0] * (n - len(l))
return r > l
def _check_update(self, force_check=False, is_manual=False):
if not UPDATE_ENABLED:
if is_manual:
self._show_toast("功能未启用")
return
if not UPDATE_API or "example.com" in UPDATE_API:
return
now = time.time()
if not force_check and now - getattr(self, "_last_update_check", 0.0) < UPDATE_INTERVAL:
return
self._last_update_check = now
def work():
try:
req = urllib.request.Request(UPDATE_API, headers={"User-Agent": UA})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=6) as r:
raw = r.read().decode("utf-8", errors="ignore")
data = json.loads(raw)
if not isinstance(data, dict):
return
remote_ver = str(data.get("version", "")).strip()
if not remote_ver:
return
if str(remote_ver).strip() == str(self.VERSION).strip():
if is_manual:
self._show_toast("当前已是最新版本 ({})".format(self.VERSION))
return
changelog = data.get("changelog") or data.get("content") or data.get("desc") or ""
force = bool(data.get("force", False))
self._show_update_dialog(remote_ver, str(changelog), force)
except Exception as e:
pass
threading.Thread(target=work).start()
def _show_update_dialog(self, remote_ver, changelog, force=False):
try:
from java import dynamic_proxy, jclass
from java.lang import Runnable
except Exception as e:
return
try:
activity = self._current_android_activity(jclass)
except Exception as e:
return
owner = self
VERSION_LOCAL = self.VERSION
REMOTE_VER = remote_ver
CHANGELOG = changelog
FORCE = force
def build_and_show():
try:
Dialog = jclass("android.app.Dialog")
LayoutParams = jclass("android.view.WindowManager$LayoutParams")
LinearLayout = jclass("android.widget.LinearLayout")
FrameLayout = jclass("android.widget.FrameLayout")
TextView = jclass("android.widget.TextView")
ScrollView = jclass("android.widget.ScrollView")
OnClickListener = jclass("android.view.View$OnClickListener")
Color = jclass("android.graphics.Color")
GD = jclass("android.graphics.drawable.GradientDrawable")
CD = jclass("android.graphics.drawable.ColorDrawable")
Gravity = jclass("android.view.Gravity")
Typeface = jclass("android.graphics.Typeface")
dm = activity.getResources().getDisplayMetrics()
density = dm.density
sh = dm.heightPixels
sw = dm.widthPixels
def dp(v):
return int(round(density * v))
COLOR_PRIMARY = Color.parseColor("#4A90E2")
COLOR_BG_SOFT = Color.parseColor("#F0F4FA")
COLOR_TEXT_MAIN = Color.parseColor("#1F2937")
COLOR_TEXT_SUB = Color.parseColor("#6B7280")
COLOR_WHITE = Color.parseColor("#FFFFFF")
def round_bg(color, radius_dp):
g = GD()
g.setShape(GD.RECTANGLE)
g.setColor(color)
g.setCornerRadius(float(dp(radius_dp)))
return g
dialog = Dialog(activity)
dialog.requestWindowFeature(1)
w = dialog.getWindow()
w.setBackgroundDrawable(CD(Color.TRANSPARENT))
w.setGravity(Gravity.CENTER)
dialog.setCanceledOnTouchOutside(not FORCE)
dialog.setCancelable(not FORCE)
card = LinearLayout(activity)
card.setOrientation(LinearLayout.VERTICAL)
card.setBackground(round_bg(COLOR_WHITE, 18))
card.setPadding(dp(22), dp(22), dp(22), dp(18))
header = LinearLayout(activity)
header.setOrientation(LinearLayout.HORIZONTAL)
header.setGravity(Gravity.CENTER_VERTICAL)
icon_wrap = FrameLayout(activity)
icon_wrap.setLayoutParams(LinearLayout.LayoutParams(dp(46), dp(46)))
icon_wrap.setBackground(round_bg(COLOR_PRIMARY, 23))
icon_tv = TextView(activity)
icon_tv.setText("↑")
icon_tv.setTextSize(22)
icon_tv.setTextColor(COLOR_WHITE)
icon_tv.setGravity(Gravity.CENTER)
icon_wrap.addView(icon_tv, FrameLayout.LayoutParams(-1, -1, Gravity.CENTER))
header.addView(icon_wrap)
title_box = LinearLayout(activity)
title_box.setOrientation(LinearLayout.VERTICAL)
lp = LinearLayout.LayoutParams(0, -2, 1.0)
lp.setMargins(dp(14), 0, 0, 0)
title_box.setLayoutParams(lp)
title_tv = TextView(activity)
title_tv.setText("发现新版本")
title_tv.setTextSize(18)
title_tv.setTextColor(COLOR_TEXT_MAIN)
try:
title_tv.setTypeface(None, Typeface.BOLD)
except Exception:
pass
title_box.addView(title_tv)
sub_tv = TextView(activity)
sub_tv.setText("新版本已就绪，建议立即更新")
sub_tv.setTextSize(12)
sub_tv.setTextColor(COLOR_TEXT_SUB)
sub_tv.setPadding(0, dp(4), 0, 0)
title_box.addView(sub_tv)
header.addView(title_box)
card.addView(header)
ver_box = LinearLayout(activity)
ver_box.setOrientation(LinearLayout.HORIZONTAL)
ver_box.setGravity(Gravity.CENTER_VERTICAL)
ver_box.setBackground(round_bg(COLOR_BG_SOFT, 12))
ver_box.setPadding(dp(16), dp(14), dp(16), dp(14))
lp = LinearLayout.LayoutParams(-1, -2)
lp.setMargins(0, dp(18), 0, 0)
ver_box.setLayoutParams(lp)
old_tv = TextView(activity)
old_tv.setText(VERSION_LOCAL)
old_tv.setTextSize(15)
old_tv.setTextColor(COLOR_TEXT_SUB)
arrow_tv = TextView(activity)
arrow_tv.setText(" → ")
arrow_tv.setTextSize(15)
arrow_tv.setTextColor(COLOR_TEXT_SUB)
new_tv = TextView(activity)
new_tv.setText(REMOTE_VER)
new_tv.setTextSize(15)
new_tv.setTextColor(COLOR_PRIMARY)
try:
new_tv.setTypeface(None, Typeface.BOLD)
except Exception:
pass
ver_box.addView(old_tv)
ver_box.addView(arrow_tv)
ver_box.addView(new_tv)
card.addView(ver_box)
sec_title = TextView(activity)
sec_title.setText("更新内容")
sec_title.setTextSize(13)
sec_title.setTextColor(COLOR_TEXT_MAIN)
try:
sec_title.setTypeface(None, Typeface.BOLD)
except Exception:
pass
sec_title.setPadding(0, dp(18), 0, dp(8))
card.addView(sec_title)
sv = ScrollView(activity)
lp = LinearLayout.LayoutParams(-1, int(sh * 0.22))
sv.setLayoutParams(lp)
sv.setBackground(round_bg(COLOR_BG_SOFT, 10))
sv.setPadding(dp(14), dp(12), dp(14), dp(12))
msg_tv = TextView(activity)
msg_tv.setText(CHANGELOG or "（无更新说明）")
msg_tv.setTextSize(13)
msg_tv.setTextColor(COLOR_TEXT_SUB)
try:
msg_tv.setLineSpacing(dp(5), 1.0)
except Exception:
pass
sv.addView(msg_tv)
card.addView(sv)
btn_box = LinearLayout(activity)
btn_box.setOrientation(LinearLayout.HORIZONTAL)
btn_box.setGravity(Gravity.CENTER_VERTICAL)
lp = LinearLayout.LayoutParams(-1, -2)
lp.setMargins(0, dp(18), 0, 0)
btn_box.setLayoutParams(lp)
if not FORCE:
close_tv = TextView(activity)
close_tv.setText("稍后再说")
close_tv.setTextSize(15)
close_tv.setTextColor(COLOR_TEXT_SUB)
close_tv.setGravity(Gravity.CENTER)
close_tv.setBackground(round_bg(COLOR_BG_SOFT, 24))
close_tv.setClickable(True)
lp = LinearLayout.LayoutParams(0, dp(46), 1.0)
lp.setMargins(0, 0, dp(6), 0)
close_tv.setLayoutParams(lp)
class CloseClick(dynamic_proxy(OnClickListener)):
def onClick(self, v):
try:
dialog.dismiss()
except Exception:
pass
close_tv.setOnClickListener(CloseClick())
btn_box.addView(close_tv)
update_tv = TextView(activity)
update_tv.setText("立即更新")
update_tv.setTextSize(15)
update_tv.setTextColor(COLOR_WHITE)
update_tv.setGravity(Gravity.CENTER)
try:
update_tv.setTypeface(None, Typeface.BOLD)
except Exception:
pass
update_tv.setBackground(round_bg(COLOR_PRIMARY, 24))
update_tv.setClickable(True)
lp = LinearLayout.LayoutParams(0, dp(46), 1.0)
lp.setMargins(dp(6), 0, 0, 0)
update_tv.setLayoutParams(lp)
class UpdateClick(dynamic_proxy(OnClickListener)):
def onClick(self, v):
try:
dialog.dismiss()
except Exception:
pass
owner._show_guide_dialog()
update_tv.setOnClickListener(UpdateClick())
btn_box.addView(update_tv)
card.addView(btn_box)
dialog.setContentView(card)
target_w = int(sw * 0.85)
max_w = dp(400)
if target_w > max_w:
target_w = max_w
w.setLayout(target_w, LayoutParams.WRAP_CONTENT)
dialog.show()
owner._dialog_refs.append(dialog)
owner._dialog_refs = owner._dialog_refs[-8:]
except Exception as e:
pass
class UIR(dynamic_proxy(Runnable)):
def run(self):
build_and_show()
activity.runOnUiThread(UIR())
def _show_guide_dialog(self):
try:
from java import dynamic_proxy, jclass
from java.lang import Runnable
except Exception as e:
return
try:
activity = self._current_android_activity(jclass)
except Exception as e:
return
owner = self
def save_bitmap_to_gallery(bitmap, filename):
try:
Bitmap_CompressFormat = jclass("android.graphics.Bitmap$CompressFormat")
MediaStoreImages = jclass("android.provider.MediaStore$Images$Media")
ContentValues = jclass("android.content.ContentValues")
Build_VERSION = jclass("android.os.Build$VERSION")
Intent = jclass("android.content.Intent")
Uri = jclass("android.net.Uri")
File = jclass("java.io.File")
FileOutputStream = jclass("java.io.FileOutputStream")
Environment = jclass("android.os.Environment")
resolver = activity.getContentResolver()
if Build_VERSION.SDK_INT >= 29:
values = ContentValues()
values.put("_display_name", filename)
values.put("mime_type", "image/png")
values.put("relative_path", "Pictures/TVBox")
uri = resolver.insert(MediaStoreImages.EXTERNAL_CONTENT_URI, values)
if uri is None:
return False
out = resolver.openOutputStream(uri)
if out is None:
return False
bitmap.compress(Bitmap_CompressFormat.PNG, 100, out)
out.flush()
out.close()
return True
else:
pics_dir = Environment.getExternalStoragePublicDirectory(
Environment.DIRECTORY_PICTURES
)
target_dir = File(pics_dir, "TVBox")
if not target_dir.exists():
target_dir.mkdirs()
target_file = File(target_dir, filename)
fos = FileOutputStream(target_file)
bitmap.compress(Bitmap_CompressFormat.PNG, 100, fos)
fos.flush()
fos.close()
try:
intent = Intent(Intent.ACTION_MEDIA_SCANNER_SCAN_FILE, Uri.fromFile(target_file))
activity.sendBroadcast(intent)
except Exception:
pass
return True
except Exception:
return False
state = {"bitmap": None, "dialog": None, "loading": None, "iv": None}
def build_and_show():
try:
Dialog = jclass("android.app.Dialog")
LayoutParams = jclass("android.view.WindowManager$LayoutParams")
LinearLayout = jclass("android.widget.LinearLayout")
TextView = jclass("android.widget.TextView")
ImageView = jclass("android.widget.ImageView")
ScaleType = jclass("android.widget.ImageView$ScaleType")
OnClickListener = jclass("android.view.View$OnClickListener")
Color = jclass("android.graphics.Color")
GD = jclass("android.graphics.drawable.GradientDrawable")
CD = jclass("android.graphics.drawable.ColorDrawable")
Gravity = jclass("android.view.Gravity")
Typeface = jclass("android.graphics.Typeface")
Toast = jclass("android.widget.Toast")
dm = activity.getResources().getDisplayMetrics()
density = dm.density
sw = dm.widthPixels
def dp(v):
return int(round(density * v))
COLOR_PRIMARY = Color.parseColor("#4A90E2")
COLOR_TEXT_MAIN = Color.parseColor("#1F2937")
COLOR_TEXT_SUB = Color.parseColor("#6B7280")
COLOR_WHITE = Color.parseColor("#FFFFFF")
def round_bg(color, radius_dp):
g = GD()
g.setShape(GD.RECTANGLE)
g.setColor(color)
g.setCornerRadius(float(dp(radius_dp)))
return g
dialog = Dialog(activity)
dialog.requestWindowFeature(1)
w = dialog.getWindow()
w.setBackgroundDrawable(CD(Color.TRANSPARENT))
w.setGravity(Gravity.CENTER)
state["dialog"] = dialog
card = LinearLayout(activity)
card.setOrientation(LinearLayout.VERTICAL)
card.setBackground(round_bg(COLOR_WHITE, 18))
card.setPadding(dp(20), dp(22), dp(20), dp(18))
title = TextView(activity)
title.setText("获取方式")
title.setTextSize(18)
title.setTextColor(COLOR_TEXT_MAIN)
title.setGravity(Gravity.CENTER)
try:
title.setTypeface(None, Typeface.BOLD)
except Exception:
pass
card.addView(title)
sub = TextView(activity)
sub.setText("微信/QQ扫码进入小程序获取")
sub.setTextSize(12)
sub.setTextColor(COLOR_TEXT_SUB)
sub.setGravity(Gravity.CENTER)
sub.setPadding(0, dp(6), 0, dp(16))
card.addView(sub)
iv = ImageView(activity)
iv.setAdjustViewBounds(True)
iv.setScaleType(ScaleType.FIT_CENTER)
lp = LinearLayout.LayoutParams(-1, -2)
iv.setLayoutParams(lp)
iv.setMinimumHeight(dp(220))
card.addView(iv)
state["iv"] = iv
loading = TextView(activity)
loading.setText("图片加载中...")
loading.setTextSize(13)
loading.setTextColor(COLOR_TEXT_SUB)
loading.setGravity(Gravity.CENTER)
loading.setPadding(0, dp(40), 0, dp(40))
card.addView(loading)
state["loading"] = loading
close_tv = TextView(activity)
close_tv.setText("保存相册")
close_tv.setTextSize(15)
close_tv.setTextColor(COLOR_WHITE)
close_tv.setGravity(Gravity.CENTER)
try:
close_tv.setTypeface(None, Typeface.BOLD)
except Exception:
pass
close_tv.setBackground(round_bg(COLOR_PRIMARY, 24))
close_tv.setClickable(True)
lp = LinearLayout.LayoutParams(-1, dp(46))
lp.setMargins(0, dp(18), 0, 0)
close_tv.setLayoutParams(lp)
class CloseClick(dynamic_proxy(OnClickListener)):
def onClick(self, v):
try:
dialog.dismiss()
except Exception:
pass
bmp = state.get("bitmap")
if bmp is None:
return
def do_save():
filename = "guide_{}.png".format(int(time.time()))
ok = save_bitmap_to_gallery(bmp, filename)
try:
if ok:
Toast.makeText(activity, "已保存到相册 Pictures/TVBox", Toast.LENGTH_LONG).show()
else:
Toast.makeText(activity, "保存失败", Toast.LENGTH_LONG).show()
except Exception:
pass
threading.Thread(target=do_save).start()
close_tv.setOnClickListener(CloseClick())
card.addView(close_tv)
dialog.setContentView(card)
target_w = int(sw * 0.82)
max_w = dp(400)
if target_w > max_w:
target_w = max_w
w.setLayout(target_w, LayoutParams.WRAP_CONTENT)
dialog.show()
except Exception as e:
pass
class UIR(dynamic_proxy(Runnable)):
def run(self):
build_and_show()
activity.runOnUiThread(UIR())
def load_image():
img_bytes = owner._get_guide_bytes()
if not img_bytes:
return
try:
BitmapFactory = jclass("android.graphics.BitmapFactory")
bitmap = BitmapFactory.decodeByteArray(img_bytes, 0, len(img_bytes))
if bitmap is None:
return
state["bitmap"] = bitmap
class SetBitmapUI(dynamic_proxy(Runnable)):
def run(self):
try:
iv = state.get("iv")
loading = state.get("loading")
if iv is not None:
iv.setImageBitmap(bitmap)
if loading is not None:
try:
parent = loading.getParent()
if parent is not None:
parent.removeView(loading)
except Exception:
pass
except Exception:
pass
activity.runOnUiThread(SetBitmapUI())
except Exception:
pass
threading.Thread(target=load_image).start()
def _load_data(self):
if self._cache is not None:
return self._cache
data = {"class": [], "list": []}
try:
r = self.session.get(DATA_API, timeout=10)
if r.status_code == 200:
j = r.json()
if isinstance(j, dict):
data["class"] = j.get("class", []) or []
data["list"] = j.get("list", []) or []
except Exception:
pass
self._cache = data
return data
def _class_name_of(self, tid):
data = self._load_data()
tid_str = str(tid)
for c in data.get("class", []) or []:
if str(c.get("type_id", "")) == tid_str:
return str(c.get("type_name", "")).strip()
return ""
def _is_last_class(self, tid):
return self._class_name_of(tid) in LAST_CLASS_NAMES
def _is_browser_only(self, name, url):
name_s = str(name or "").strip()
url_s = str(url or "").strip()
if name_s in BROWSER_ONLY_NAMES:
return True
for key in BROWSER_ONLY_URL_KEYS:
if key and key in url_s:
return True
return False
def _web_vod(self, item):
vid = str(item.get("vod_id", ""))
web_url = item.get("web_url", "")
vod = {
"vod_id": vid,
"vod_name": item.get("vod_name", ""),
"vod_pic": item.get("vod_pic", ""),
"vod_remarks": item.get("vod_remarks", ""),
}
if vid == "9":
vod["action"] = self.ACTION_SEND
return vod
if vid == "10":
vod["action"] = self.ACTION_RECV
return vod
if vid and web_url:
if self._is_browser_only(vod["vod_name"], web_url):
vod["action"] = self.ACTION_BROWSER_OPEN_PREFIX + vid
else:
vod["action"] = self.ACTION_WEB_OPEN_PREFIX + vid
return vod
def _last_class_vod(self, tid):
return {
"vod_id": self.CHECK_UPDATE_ID,
"vod_name": "检查更新",
"vod_pic": "https://raw.gitcode.com/u013624795/jk/raw/main/gx.jpg",
"vod_remarks": "版本: {} · 点击检查更新".format(self.VERSION),
"action": self.ACTION_CHECK_UPDATE,
}
def homeContent(self, filter):
self._ensure_inited()
data = self._load_data()
normal_classes = []
last_classes = []
for c in data.get("class", []) or []:
name = str(c.get("type_name", "")).strip()
if name in LAST_CLASS_NAMES:
last_classes.append(c)
else:
normal_classes.append(c)
classes = normal_classes
classes.append({"type_id": self.TID_PY_OPS, "type_name": "OK推送PY"})
classes.append({
"type_id": self.TID_PY_LIST,
"type_name": "PY源({})".format(len(self.py_sources)),
})
for c in last_classes:
nc = dict(c)
nc["type_name"] = "关于"
classes.append(nc)
return {"class": classes}
def categoryContent(self, tid, pg, filter, extend):
self._ensure_inited()
tid_str = str(tid)
if tid_str == self.TID_PY_OPS:
ok_count = 0
for s in self.py_sources:
r = self.check_results.get(self._source_hash(s))
if r and r.get("ok"):
ok_count += 1
if self.check_state["status"] == "running":
check_remarks = "检测中 {}/{}".format(self.check_state["done"], self.check_state["total"])
report_remarks = "检测仍在进行中…"
elif self.check_state["status"] == "done":
bad_count = len(self.py_sources) - ok_count
check_remarks = "重新检测"
report_remarks = "可用 {}，废弃 {}".format(ok_count, bad_count)
else:
check_remarks = "开始检测"
report_remarks = "暂无历史数据"
return {
"page": 1, "pagecount": 1, "limit": 20, "total": 4,
"list": [
{"vod_id": self.RESCAN_ID, "vod_name": "扫描推送", "vod_pic": PY_OPS_PIC,
"vod_remarks": "已扫 {} 个 · 点击直接执行".format(len(self.py_sources)),
"action": self.ACTION_RESCAN},
{"vod_id": self.EDIT_DIR_ID, "vod_name": "设置目录", "vod_pic": PY_OPS_PIC,
"vod_remarks": "当前: {}".format(self.scan_dir),
"action": self.ACTION_EDIT_DIR},
{"vod_id": self.CHECK_ALL_ID, "vod_name": "检测 PY 源", "vod_pic": PY_OPS_PIC,
"vod_remarks": check_remarks, "action": self.ACTION_CHECK_ALL},
{"vod_id": self.VIEW_REPORT_ID, "vod_name": "查看检测报告", "vod_pic": PY_OPS_PIC,
"vod_remarks": report_remarks, "action": self.ACTION_VIEW_REPORT},
],
}
if tid_str == self.TID_PY_LIST:
try:
page = max(1, int(pg))
except Exception:
page = 1
return self._paged([self._to_py_vod(s) for s in self.py_sources], page)
if self._is_last_class(tid_str):
data = self._load_data()
vod_list = [self._last_class_vod(tid_str)]
for item in data.get("list", []) or []:
if str(item.get("type_id", "")) != tid_str:
continue
vod_list.append(self._web_vod(item))
return {"page": 1, "pagecount": 1, "limit": 20,
"total": len(vod_list), "list": vod_list}
data = self._load_data()
vod_list = []
for item in data.get("list", []):
if str(item.get("type_id", "")) != tid_str:
continue
vod_list.append(self._web_vod(item))
return {"list": vod_list, "page": 1, "pagecount": 1,
"limit": 20, "total": len(vod_list)}
def _paged(self, items, page):
page_size = 60
total = len(items)
page_count = max(1, (total + page_size - 1) // page_size)
if page > page_count:
page_items = []
else:
start = (page - 1) * page_size
page_items = items[start:start + page_size]
return {"page": page, "pagecount": page_count,
"limit": page_size, "total": total, "list": page_items}
def _to_py_vod(self, source):
h = self._source_hash(source)
r = self.check_results.get(h)
if r is None:
status = "【未测】"
elif r.get("ok"):
status = "【可用】"
elif r.get("timeout"):
status = "【超时】"
else:
status = "【废源】"
err_msg = str(r.get("message", "")).replace("\n", " ")
if len(err_msg) > 15:
err_msg = err_msg[:15] + "…"
status += " " + err_msg
return {
"vod_id": self._source_id(source),
"vod_name": source["name"],
"vod_pic": PY_SOURCE_PIC,
"vod_remarks": "{} {}".format(status, "直接推送"),
"action": self.ACTION_PUSH_ONE_PREFIX + h,
}
def searchContent(self, key, quick, pg=1):
self._ensure_inited()
keyword = str(key or "").strip()
kw_low = keyword.lower()
data = self._load_data()
vod_list = []
for item in data.get("list", []):
name = item.get("vod_name", "")
if keyword and keyword not in name:
continue
vod_list.append(self._web_vod(item))
if kw_low:
for s in self.py_sources:
if kw_low in s["name"].lower() or kw_low in s["relative"].lower():
vod_list.append(self._to_py_vod(s))
return {"list": vod_list, "page": 1, "pagecount": 1,
"limit": 20, "total": len(vod_list)}
def detailContent(self, ids):
self._ensure_inited()
vid = str(ids[0]) if isinstance(ids, list) and ids else str(ids)
if vid in (self.CHECK_ALL_ID, self.VIEW_REPORT_ID):
return {"list": [{"vod_id": vid, "vod_name": "检测 PY 源",
"vod_pic": PY_OPS_PIC, "vod_remarks": "请从列表按钮点击",
"vod_content": "请返回列表点击按钮"}]}
if vid == self.RESCAN_ID:
ok_scan, ok_push, push_msg = self._do_rescan_and_push()
return {"list": [{"vod_id": vid, "vod_name": "扫描并推送完成",
"vod_pic": PY_OPS_PIC,
"vod_remarks": "成功" if ok_scan and ok_push else "未推送",
"vod_content": "扫描: {}\n扫到: {} 个PY源\n推送: {}".format(
"成功" if ok_scan else "失败", len(self.py_sources), push_msg)}]}
if vid == self.EDIT_DIR_ID:
opened, message = self._open_dir_browser()
return {"list": [{"vod_id": vid, "vod_name": "设置目录",
"vod_pic": PY_OPS_PIC,
"vod_remarks": "已打开" if opened else "失败",
"vod_content": message or "已打开"}]}
if vid == self.CHECK_UPDATE_ID:
self._last_update_check = 0
self._check_update(force_check=True, is_manual=True)
return {"list": [{"vod_id": vid, "vod_name": "手动更新",
"vod_pic": "https://raw.gitcode.com/u013624795/jk/raw/main/gx.jpg",
"vod_remarks": "已发起", "vod_content": "当前版本: " + self.VERSION}]}
src = self._find_py_source_by_id(vid)
if src is not None:
with self.lock:
ok, msg = self._push_one(self._source_hash(src))
return {"list": [{"vod_id": vid, "vod_name": src["name"],
"vod_pic": PY_SOURCE_PIC,
"vod_remarks": "推送" + ("成功" if ok else "失败"),
"vod_content": "文件: {}\n\n{}".format(src["path"], msg)}]}
data = self._load_data()
for item in data.get("list", []):
if str(item.get("vod_id", "")) == vid:
web_url = item.get("web_url", "")
vod_name = item.get("vod_name", "")
self._cur_url = web_url
self._cur_name = vod_name
if self._is_browser_only(vod_name, web_url):
self._open_in_browser(web_url)
return {"list": [{"vod_id": vid, "vod_name": vod_name,
"vod_pic": item.get("vod_pic", ""),
"vod_content": item.get("vod_content", "") or "",
"vod_play_from": "源码分享QQ交流群:212706934",
"vod_play_url": f"资源共享https://fzl.xo.je${web_url}"}]}
return {"list": []}
def playerContent(self, flag, id, vipFlags):
if id and id.startswith("http"):
if self._is_browser_only(self._cur_name, id):
self._open_in_browser(id)
return {"parse": 1, "url": id, "header": "{}"}
self._show_web_dialog(id)
return {"parse": 1, "url": id, "header": "{}"}
return {"parse": 0, "url": "", "header": {}, "msg": "无播放内容"}
def action(self, action):
self._ensure_inited()
action = str(action)
if action == self.ACTION_SEND:
self._show_send_file_dialog()
return {"code": 0, "msg": ""}
if action == self.ACTION_RECV:
self._show_recv_file_dialog()
return {"code": 0, "msg": ""}
if action == self.ACTION_CHECK_ALL or action == self.ACTION_VIEW_REPORT:
opened, message = self._open_check_dialog(force_restart=False)
return {"code": 0, "msg": "已打开" if opened else message}
if action.startswith(self.ACTION_CHECK_ONE_PREFIX):
h = action[len(self.ACTION_CHECK_ONE_PREFIX):]
source = self._find_source_by_hash(h)
if source is None:
return {"code": 0, "msg": "未找到该PY源"}
with self.lock:
result = self._check_source(source)
self.check_results[h] = result
self._save_check_results()
prefix = "✓ 可用" if result.get("ok") else ("⏳ 超时" if result.get("timeout") else "✗ 废源")
return {"code": 0, "msg": "{}：{}".format(prefix, result.get("message", ""))}
if action.startswith(self.ACTION_BROWSER_OPEN_PREFIX):
vid = action[len(self.ACTION_BROWSER_OPEN_PREFIX):]
data = self._load_data()
for item in data.get("list", []):
if str(item.get("vod_id", "")) == vid:
wu = item.get("web_url", "")
if not wu:
return {"code": 0, "msg": ""}
self._cur_url = wu
self._cur_name = item.get("vod_name", "")
self._open_in_browser(wu)
return {"code": 0, "msg": ""}
return {"code": 0, "msg": ""}
if action.startswith(self.ACTION_WEB_OPEN_PREFIX):
vid = action[len(self.ACTION_WEB_OPEN_PREFIX):]
data = self._load_data()
for item in data.get("list", []):
if str(item.get("vod_id", "")) == vid:
wu = item.get("web_url", "")
if not wu:
return {"code": 0, "msg": ""}
self._cur_url = wu
self._cur_name = item.get("vod_name", "")
self._show_web_dialog(wu)
return {"code": 0, "msg": ""}
return {"code": 0, "msg": ""}
if action == self.ACTION_RESCAN:
ok_scan, ok_push, push_msg = self._do_rescan_and_push()
if ok_scan and self.py_sources and ok_push:
return {"code": 0, "msg": "扫描完成：{} 个\n{}".format(len(self.py_sources), push_msg)}
if ok_scan and not self.py_sources:
return {"code": 0, "msg": "扫描完成：0 个"}
return {"code": 0, "msg": "扫描结果: {}".format(push_msg)}
if action == self.ACTION_EDIT_DIR:
opened, message = self._open_dir_browser()
return {"code": 0, "msg": "" if opened else message}
if action == self.ACTION_CHECK_UPDATE:
self._last_update_check = 0
self._check_update(force_check=True, is_manual=True)
return {"code": 0, "msg": ""}
if action.startswith(self.ACTION_PUSH_ONE_PREFIX):
h = action[len(self.ACTION_PUSH_ONE_PREFIX):]
with self.lock:
ok, msg = self._push_one(h)
return {"code": 0, "msg": msg}
return {"code": 0, "msg": ""}
def _do_rescan_and_push(self):
with self.lock:
ok_scan = self._refresh()
if ok_scan and self.py_sources:
ok_push, push_msg = self._push_all()
elif ok_scan:
ok_push, push_msg = False, "扫到 0 个"
else:
ok_push, push_msg = False, "扫描失败"
return ok_scan, ok_push, push_msg
def _load_check_results(self):
if not os.path.isfile(CHECK_RESULTS_PATH):
return
try:
with open(CHECK_RESULTS_PATH, "r", encoding="utf-8") as fp:
data = json.load(fp)
if isinstance(data, dict):
self.check_results = data
except Exception:
self.check_results = {}
def _save_check_results(self):
try:
if not os.path.isdir(TV_DIR):
os.makedirs(TV_DIR, exist_ok=True)
temp_path = CHECK_RESULTS_PATH + ".tmp"
with open(temp_path, "w", encoding="utf-8") as fp:
json.dump(self.check_results, fp, ensure_ascii=False, indent=2)
fp.flush()
os.fsync(fp.fileno())
os.replace(temp_path, CHECK_RESULTS_PATH)
except Exception:
pass
def _check_source(self, source):
holder = {"ok": False, "message": "", "timeout": False}
path = source["path"]
module_name = "_pycheck_{}".format(self._source_hash(source))
def worker():
try:
with open(path, "rb") as fp:
raw = fp.read(self.MAX_SOURCE_SIZE)
code = compile(raw, path, "exec")
except SyntaxError as exc:
holder["message"] = "语法错误 line {}".format(getattr(exc, "lineno", "?"))
return
except BaseException:
holder["message"] = "读取失败"
return
namespace = {
"__name__": module_name,
"__file__": path,
"__package__": "",
"__doc__": None,
}
old_path = list(sys.path)
file_dir = os.path.dirname(path)
if file_dir not in sys.path:
sys.path.insert(0, file_dir)
if self.scan_dir not in sys.path:
sys.path.insert(0, self.scan_dir)
try:
exec(code, namespace)
except ImportError as exc:
holder["message"] = "缺少依赖: {}".format(exc).split('.')[-1]
return
except BaseException:
holder["message"] = "顶层执行失败"
return
finally:
sys.path = old_path
if module_name in sys.modules:
del sys.modules[module_name]
spider_class = None
candidate = namespace.get("Spider")
if isinstance(candidate, type):
spider_class = candidate
if spider_class is None:
try:
for _, value in list(namespace.items()):
if isinstance(value, type) and issubclass(value, BaseSpider) and value is not BaseSpider:
spider_class = value
break
except BaseException:
pass
if spider_class is None:
holder["message"] = "未找到 Spider 类"
return
try:
instance = spider_class()
instance.init("")
except BaseException as exc:
holder["message"] = "初始化失败: {}".format(type(exc).__name__)
return
home_class_n = 0
home = None
try:
home = instance.homeContent(False)
if isinstance(home, dict):
home_list_n = len(home.get("list") or [])
home_class_n = len(home.get("class") or [])
if home_list_n > 0:
holder["ok"] = True
holder["message"] = "可用 (首页有数据)"
return
else:
holder["message"] = "首页返回格式错误"
return
except BaseException as exc:
holder["message"] = "首页加载异常"
return
if home_class_n == 0:
holder["message"] = "无分类，直接废弃"
return
try:
tid = home["class"][0]["type_id"]
cat = instance.categoryContent(tid, "1", False, {})
if isinstance(cat, dict) and len(cat.get("list") or []) > 0:
holder["ok"] = True
holder["message"] = "可用 (分类有数据)"
return
except BaseException:
pass
try:
search = instance.searchContent("斗罗", False, "1")
if isinstance(search, dict) and len(search.get("list") or []) > 0:
holder["ok"] = True
holder["message"] = "可用 (搜索有数据)"
return
except BaseException:
pass
holder["message"] = "分类与搜索均无数据"
t = threading.Thread(target=worker, daemon=True)
t.start()
t.join(self.CHECK_TIMEOUT)
result = {
"ok": holder["ok"],
"message": holder["message"],
"time": time.strftime("%Y-%m-%d %H:%M:%S"),
"ts": time.time(),
"timeout": holder["timeout"]
}
if t.is_alive():
result["ok"] = False
result["timeout"] = True
result["message"] = "网络超时"
return result
def _open_check_dialog(self, force_restart=False):
try:
from java import dynamic_proxy, jclass
toast_class = jclass("android.widget.Toast")
runnable_class = jclass("java.lang.Runnable")
click_listener_class = jclass("android.content.DialogInterface$OnClickListener")
dismiss_listener_class = jclass("android.content.DialogInterface$OnDismissListener")
try:
builder_class = jclass("com.google.android.material.dialog.MaterialAlertDialogBuilder")
except Exception:
builder_class = jclass("android.app.AlertDialog$Builder")
activity = self._current_android_activity(jclass)
owner = self
state = owner.check_state
if force_restart:
state["status"] = "idle"
state["results"] = []
state["total"] = 0
state["done"] = 0
def build_text():
lines = []
total = state["total"]
done = state["done"]
if state["status"] == "running":
lines.append("进度: {}/{} 深度检测中…".format(done, total))
if state["current"]:
lines.append("当前: {}".format(state["current"]))
lines.append("")
recent = state["results"][-8:]
for r in recent:
ok_flag = r["ok"]
if r.get("timeout"): mark = "⏳"
elif ok_flag: mark = "✓"
else: mark = "✗"
short_name = r["name"] if len(r["name"]) <= 15 else r["name"][:15] + "…"
short_msg = r["msg"] if len(r["msg"]) <= 25 else r["msg"][:25] + "…"
lines.append("{} {}: {}".format(mark, short_name, short_msg))
elif state["status"] == "done":
if total == 0:
lines.append("扫描结果为空，未检测")
else:
ok_items = [r for r in state["results"] if r["ok"]]
timeout_items = [r for r in state["results"] if r.get("timeout") and not r["ok"]]
bad_items = [r for r in state["results"] if not r["ok"] and not r.get("timeout")]
lines.append("共检测 {} 个：可用 {}，废弃 {}，超时 {}".format(
len(state["results"]), len(ok_items), len(bad_items), len(timeout_items)))
if bad_items:
lines.append("")
lines.append("【废弃/无数据源】:")
for r in bad_items:
short_name = r["name"] if len(r["name"]) <= 20 else r["name"][:20] + "…"
short_msg = r["msg"] if len(r["msg"]) <= 30 else r["msg"][:30] + "…"
lines.append("✗ {}: {}".format(short_name, short_msg))
if timeout_items:
lines.append("")
lines.append("【网络超时源】(已复测，仍超时，建议手动测试):")
for r in timeout_items:
short_name = r["name"] if len(r["name"]) <= 20 else r["name"][:20] + "…"
lines.append("⏳ {}".format(short_name))
if not bad_items and not timeout_items:
lines.append("")
lines.append("🎉 所有源检测均有真实数据！")
else:
lines.append("尚未开始检测。")
return "\n".join(lines)
def update_dialog():
d = state.get("dialog")
if d is None:
return
try:
d.setMessage(build_text())
except BaseException:
pass
class UIUpdateRunner(dynamic_proxy(runnable_class)):
def __init__(self):
super().__init__()
self.task = None
def run(self):
if self.task:
try:
self.task()
except BaseException:
pass
def post_ui(fn):
try:
runner = UIUpdateRunner()
runner.task = fn
activity.runOnUiThread(runner)
except BaseException:
pass
class MyDismissListener(dynamic_proxy(dismiss_listener_class)):
def onDismiss(self, dialog):
if state.get("dialog") == dialog:
state["dialog"] = None
class ReCheckButton(dynamic_proxy(click_listener_class)):
def onClick(self, dialog, which):
dialog.dismiss()
owner._open_check_dialog(force_restart=True)
def check_worker():
try:
with owner.lock:
sources = list(owner.py_sources)
state["total"] = len(sources)
state["done"] = 0
state["results"] = []
state["status"] = "running"
post_ui(update_dialog)
if not sources:
state["status"] = "done"
post_ui(update_dialog)
return
timeout_sources = []
for idx, source in enumerate(sources):
if state["status"] != "running":
break
state["current"] = source["name"]
post_ui(update_dialog)
result = owner._check_source(source)
h = owner._source_hash(source)
if result.get("timeout"):
timeout_sources.append(source)
state["results"].append({
"ok": False,
"name": source["name"],
"msg": "网络超时(待复测)",
"timeout": True
})
else:
state["results"].append({
"ok": result["ok"],
"name": source["name"],
"msg": result.get("message", ""),
"timeout": False
})
with owner.lock:
owner.check_results[h] = result
owner._save_check_results()
state["done"] = idx + 1
state["current"] = ""
post_ui(update_dialog)
time.sleep(0.02)
if timeout_sources and state["status"] == "running":
state["current"] = "正在复测网络超时源..."
post_ui(update_dialog)
time.sleep(0.1)
for source in timeout_sources:
if state["status"] != "running":
break
state["current"] = "复测: {}".format(source["name"])
post_ui(update_dialog)
result = owner._check_source(source)
h = owner._source_hash(source)
for r in state["results"]:
if r["name"] == source["name"]:
r["ok"] = result["ok"]
r["msg"] = result.get("message", "")
r["timeout"] = result.get("timeout", False)
break
with owner.lock:
owner.check_results[h] = result
owner._save_check_results()
state["current"] = ""
post_ui(update_dialog)
time.sleep(0.02)
except BaseException as exc:
state["results"].append({
"ok": False, "name": "内部线程崩溃", "msg": str(exc), "timeout": False
})
finally:
state["status"] = "done"
state["current"] = ""
post_ui(update_dialog)
class ShowRunner(dynamic_proxy(runnable_class)):
def run(self):
try:
b = builder_class(activity)
if state["status"] == "running":
b.setTitle("正在深度检测 PY 源")
b.setMessage(build_text())
b.setPositiveButton("关闭", None)
elif state["status"] == "done":
b.setTitle("检测结果报告")
b.setMessage(build_text())
b.setPositiveButton("关闭", None)
b.setNegativeButton("重新检测", ReCheckButton())
else:
b.setTitle("开始检测")
b.setMessage("准备开始检测…")
b.setPositiveButton("关闭", None)
b.setCancelable(True)
d = b.show()
d.setOnDismissListener(MyDismissListener())
state["dialog"] = d
owner._dialog_refs.append(d)
owner._dialog_refs = owner._dialog_refs[-8:]
if state["status"] == "idle":
state["status"] = "running"
threading.Thread(target=check_worker, daemon=True).start()
except BaseException as exc:
state["status"] = "done"
try:
toast_class.makeText(activity, "打开检测窗口失败: " + str(exc), toast_class.LENGTH_LONG).show()
except BaseException:
pass
activity.runOnUiThread(ShowRunner())
return True, ""
except BaseException as exc:
return False, "打开检测窗口失败: {}".format(exc)
def _empty_py_status(self):
return {
"scan_time": "-", "found": 0, "included": 0, "skipped": 0,
"write_state": "尚未扫描", "error": "", "warnings": [],
}
def _load_settings(self):
if not os.path.isfile(SETTINGS_PATH):
return
try:
with open(SETTINGS_PATH, "r", encoding="utf-8") as fp:
data = json.load(fp)
if isinstance(data, dict):
value = str(data.get("scan_dir", "")).strip()
if value:
self.scan_dir = value
except Exception:
pass
def _save_settings(self):
try:
if not os.path.isdir(TV_DIR):
os.makedirs(TV_DIR, exist_ok=True)
with open(SETTINGS_PATH, "w", encoding="utf-8") as fp:
json.dump({"scan_dir": self.scan_dir}, fp, ensure_ascii=False, indent=2)
except Exception:
pass
def _is_spider_file(self, path):
try:
with open(path, "rb") as fp:
data = fp.read(256 * 1024)
text = data.decode("utf-8", errors="ignore")
except Exception:
return False
return bool(re.search(r"\bclass\s+Spider\s*(?:\(|:)", text))
def _scan(self):
self.py_sources = []
self._py_status = self._empty_py_status()
self._py_status["scan_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
root = os.path.abspath(os.path.expanduser(self.scan_dir))
if not os.path.isdir(root):
self._py_status["write_state"] = "目录不存在"
self._py_status["error"] = "扫描目录不存在: {}".format(root)
return False
try:
self_path = os.path.realpath(__file__)
except Exception:
self_path = ""
found = included = skipped = 0
for current, dirs, files in os.walk(root, topdown=True, followlinks=False):
rel = os.path.relpath(current, root)
depth = 0 if rel == "." else rel.count(os.sep) + 1
dirs[:] = sorted(
[d for d in dirs if not d.startswith(".") and d != "__pycache__"],
key=lambda v: v.lower(),
)
if depth >= self.MAX_SCAN_DEPTH:
dirs[:] = []
for file_name in sorted(files, key=lambda v: v.lower()):
if not file_name.lower().endswith(".py"):
continue
if file_name == "__init__.py":
continue
if found >= self.MAX_SCAN_FILES:
break
full_path = os.path.join(current, file_name)
if os.path.islink(full_path) or not os.path.isfile(full_path):
continue
real_path = os.path.realpath(full_path)
if self_path and real_path == self_path:
continue
found += 1
try:
stat = os.stat(real_path)
if stat.st_size <= 0 or stat.st_size > self.MAX_SOURCE_SIZE:
skipped += 1
continue
except Exception:
skipped += 1
continue
if not self._is_spider_file(real_path):
skipped += 1
continue
relative = os.path.relpath(real_path, root).replace(os.sep, "/")
self.py_sources.append({
"path": real_path,
"relative": relative,
"base_name": file_name[:-3],
})
included += 1
self._apply_display_names()
self._py_status["found"] = found
self._py_status["included"] = included
self._py_status["skipped"] = skipped
self._py_status["write_state"] = "扫描完成"
return True
def _apply_display_names(self):
counts = {}
for source in self.py_sources:
key = source["base_name"].lower()
counts[key] = counts.get(key, 0) + 1
for source in self.py_sources:
base = source["base_name"]
if counts.get(base.lower(), 0) > 1:
folder = os.path.dirname(source["relative"])
suffix = folder.replace("/", " · ") if folder else "根目录"
source["name"] = "{} 【{}】┃[py]".format(base, suffix)
else:
source["name"] = base + "┃[py]"
def _file_url(self, path):
absolute = os.path.realpath(os.path.abspath(path))
storage_root = os.path.realpath(STORAGE_ROOT)
try:
relative = os.path.relpath(absolute, storage_root).replace(os.sep, "/")
except Exception:
relative = ""
if relative and relative != ".." and not relative.startswith("../"):
return "file://" + relative.lstrip("/")
return "file://" + absolute
def _abs_file_url(self, path):
return "file://" + os.path.realpath(os.path.abspath(path))
def _source_hash(self, source):
return hashlib.sha256(source["path"].encode("utf-8")).hexdigest()[:12]
def _source_id(self, source):
return "py_src_" + hashlib.sha256(source["path"].encode("utf-8")).hexdigest()[:16]
def _find_source_by_hash(self, h):
for s in self.py_sources:
if self._source_hash(s) == h:
return s
return None
def _find_py_source_by_id(self, sid):
for s in self.py_sources:
if self._source_id(s) == sid:
return s
return None
def _build_site(self, source):
return {
"key": "local_py_" + self._source_hash(source),
"name": source["name"],
"type": 3,
"api": self._file_url(source["path"]),
"searchable": 1,
"quickSearch": 1,
}
def _write_config_file(self, path, config):
try:
if not os.path.isdir(os.path.dirname(path)):
os.makedirs(os.path.dirname(path), exist_ok=True)
content = json.dumps(config, ensure_ascii=False, indent=2) + "\n"
if os.path.isfile(path):
try:
with open(path, "r", encoding="utf-8") as fp:
if fp.read() == content:
return True
except Exception:
pass
temp_path = path + ".tmp"
with open(temp_path, "w", encoding="utf-8") as fp:
fp.write(content); fp.flush(); os.fsync(fp.fileno())
os.replace(temp_path, path)
return True
except Exception as exc:
self._py_status["error"] = "写入配置失败: {}".format(exc)
return False
def _write_fengzhiling(self):
if not self.py_sources:
self._py_status["error"] = "未扫到任何 PY 源"
return False
return self._write_config_file(
FENGLING_PATH, {"sites": [self._build_site(s) for s in self.py_sources]}
)
def _write_single_config(self, source):
return self._write_config_file(SINGLE_PATH, {"sites": [self._build_site(source)]})
def _push_config(self, url, name="推送接口"):
try:
from java import jclass
port = int(jclass("com.github.catvod.Proxy").getPort())
if port < 1:
return False, "Proxy 端口未就绪"
config = json.dumps(
{"type": 0, "url": url, "name": name},
ensure_ascii=False, separators=(",", ":"),
)
body = urllib.parse.urlencode(
{"config": config, "targets": "[]", "force": "false"}
).encode("utf-8")
req = urllib.request.Request(
"http://127.0.0.1:{}/action?do=sync&mode=1&type=history".format(port),
data=body,
headers={
"Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
"Connection": "close",
},
)
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=5) as r:
code = r.getcode()
return True, "已推送 (HTTP {})".format(code)
except Exception as e:
return False, "推送失败: {}".format(e)
def _push_all(self):
if not self.py_sources:
return False, "没有扫描到py爬虫，请重新设置目录"
if not os.path.isfile(FENGLING_PATH):
return False, "爬虫不存在"
return self._push_config(self._abs_file_url(FENGLING_PATH), name="全部PY源")
def _push_one(self, h):
source = self._find_source_by_hash(h)
if source is None:
return False, "未找到该PY源"
if not self._write_single_config(source):
return False, "临时配置失败: {}".format(self._py_status.get("error", ""))
return self._push_config(self._abs_file_url(SINGLE_PATH), name=source["name"])
def _refresh(self):
ok = self._scan()
if ok and self.py_sources:
self._write_fengzhiling()
return ok
def _current_android_activity(self, jclass):
app_class = jclass("com.fongmi.android.tv.App")
activity_class = jclass("android.app.Activity")
modifier_class = jclass("java.lang.reflect.Modifier")
app_info = app_class.getClass()
activity_info = activity_class.getClass()
for method in app_info.getDeclaredMethods():
try:
if not modifier_class.isStatic(method.getModifiers()):
continue
if len(method.getParameterTypes()) != 0:
continue
if not activity_info.isAssignableFrom(method.getReturnType()):
continue
method.setAccessible(True)
try:
activity = method.invoke(None, [])
except Exception:
activity = method.invoke(None)
if activity is not None:
return activity
except Exception:
continue
app = None
for field in app_info.getDeclaredFields():
try:
if not modifier_class.isStatic(field.getModifiers()):
continue
if not app_info.isAssignableFrom(field.getType()):
continue
field.setAccessible(True)
app = field.get(None)
if app is not None:
break
except Exception:
continue
if app is not None:
for field in app.getClass().getDeclaredFields():
try:
if modifier_class.isStatic(field.getModifiers()):
continue
if not activity_info.isAssignableFrom(field.getType()):
continue
field.setAccessible(True)
activity = field.get(app)
if activity is not None:
return activity
except Exception:
continue
raise ValueError("未找到当前 Android 页面")
def _open_in_browser(self, url):
url = str(url or "").strip()
if not url:
return False
try:
from java import jclass
activity = self._current_android_activity(jclass)
Intent = jclass("android.content.Intent")
Uri = jclass("android.net.Uri")
intent = Intent(Intent.ACTION_VIEW)
intent.setData(Uri.parse(url))
intent.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
activity.startActivity(intent)
return True
except Exception as e:
self._show_toast("打开浏览器失败: {}".format(e))
return False
def _open_dir_browser(self):
try:
from java import dynamic_proxy, jclass, jarray
from java.lang import String as JString
toast_class = jclass("android.widget.Toast")
click_listener = jclass("android.content.DialogInterface$OnClickListener")
runnable_class = jclass("java.lang.Runnable")
try:
builder_class = jclass("com.google.android.material.dialog.MaterialAlertDialogBuilder")
except Exception:
builder_class = jclass("android.app.AlertDialog$Builder")
activity = self._current_android_activity(jclass)
owner = self
start = os.path.abspath(os.path.expanduser(owner.scan_dir or DEFAULT_PY_DIR))
if not os.path.isdir(start):
start = os.path.realpath(STORAGE_ROOT)
state = {"path": start, "showing": False, "dialog": None}
def list_dirs(path):
try:
return sorted([d for d in os.listdir(path)
if os.path.isdir(os.path.join(path, d)) and not d.startswith(".")],
key=lambda v: v.lower())
except Exception:
return []
def dismiss_current():
d = state.get("dialog")
if d is not None:
try:
d.dismiss()
except Exception:
pass
state["dialog"] = None
def schedule_show():
if state["showing"]:
return
state["showing"] = True
activity.runOnUiThread(ShowRunner())
def navigate_to(new_path):
state["path"] = new_path
dismiss_current()
schedule_show()
def use_current():
path = state["path"]
with owner.lock:
owner.scan_dir = path
owner._save_settings()
try:
owner._refresh()
except Exception as exc:
toast_class.makeText(activity, "扫描失败: " + str(exc),
toast_class.LENGTH_LONG).show()
return
dismiss_current()
toast_class.makeText(activity, "已扫描: {} 个".format(len(owner.py_sources)),
toast_class.LENGTH_LONG).show()
class ItemClick(dynamic_proxy(click_listener)):
def onClick(self, dialog, which):
try:
idx = int(which)
path = state["path"]
dirs = list_dirs(path)
has_parent = os.path.dirname(path) != path
offset = 0
if has_parent:
if idx == 0:
navigate_to(os.path.dirname(path))
return
offset = 1
di = idx - offset
if 0 <= di < len(dirs):
navigate_to(os.path.join(path, dirs[di]))
except Exception:
pass
class UseButton(dynamic_proxy(click_listener)):
def onClick(self, dialog, which):
try:
use_current()
except Exception:
pass
class ShowRunner(dynamic_proxy(runnable_class)):
def run(self):
try:
path = state["path"]
dirs = list_dirs(path)
labels = []
if os.path.dirname(path) != path:
labels.append("⬆ 返回上级")
if not dirs:
labels.append("（无子文件夹）")
else:
for d in dirs:
labels.append("📁 " + d)
arr = jarray(JString)(labels)
b = builder_class(activity)
b.setTitle(path)
b.setItems(arr, ItemClick())
b.setPositiveButton("使用此目录", UseButton())
b.setNegativeButton("取消", None)
d = b.show()
state["dialog"] = d
owner._dialog_refs.append(d)
owner._dialog_refs = owner._dialog_refs[-8:]
except Exception:
pass
finally:
state["showing"] = False
schedule_show()
return True, ""
except Exception as exc:
return False, str(exc)
def _read_clip_and_push(self):
def work():
try:
from java import jclass
ctx = jclass("android.app.ActivityThread").currentApplication()
cm = ctx.getSystemService(ctx.CLIPBOARD_SERVICE)
clip = cm.getPrimaryClip()
if not clip or clip.getItemCount() < 1:
return
text = str(clip.getItemAt(0).coerceToText(ctx)).strip()
try:
ClipData = jclass("android.content.ClipData")
cm.setPrimaryClip(ClipData.newPlainText("", ""))
except Exception:
pass
if text.startswith(PLAY_FLAG):
raw = text[len(PLAY_FLAG):].strip()
if "||" in raw:
url, title = raw.split("||", 1)
else:
url, title = raw.strip(), "播放"
if url.startswith(("http://", "https://")):
self._play_url_native(url.strip(), title.strip())
return
if not text.startswith(PUSH_FLAG):
return
url = text[len(PUSH_FLAG):].strip()
if url.startswith(("http://", "https://")):
self._push_config(url, name="推送")
except Exception:
pass
threading.Thread(target=work).start()
def _play_url_native(self, url, title="播放"):
try:
from java import jclass
except Exception:
return False
try:
activity = self._current_android_activity(jclass)
except Exception:
return False
Intent = jclass("android.content.Intent")
Uri = jclass("android.net.Uri")
try:
intent = Intent(Intent.ACTION_SEND)
intent.setType("text/plain")
intent.putExtra(Intent.EXTRA_TEXT, url)
try:
intent.setPackage(activity.getPackageName())
except Exception:
pass
activity.startActivity(intent)
return True
except Exception:
pass
try:
vi = Intent(Intent.ACTION_VIEW)
vi.setDataAndType(Uri.parse(url), "video/*")
activity.startActivity(vi)
return True
except Exception:
pass
return False
def _show_web_dialog(self, url=""):
title_text = self._cur_name or "网页浏览"
def launch():
try:
from java import jclass, dynamic_proxy
from java.lang import Runnable
except Exception:
return
try:
JClass = jclass("java.lang.Class")
AT = JClass.forName("android.app.ActivityThread")
currentAT = AT.getMethod("currentActivityThread").invoke(None)
mActivities = AT.getDeclaredField("mActivities")
mActivities.setAccessible(True)
values = mActivities.get(currentAT).values()
try:
records = values.toArray()
except Exception:
records = values.getClass().getMethod("toArray").invoke(values)
act = None
for record in records:
try:
rc = record.getClass()
pf = rc.getDeclaredField("paused")
pf.setAccessible(True)
if not pf.getBoolean(record):
af = rc.getDeclaredField("activity")
af.setAccessible(True)
act = af.get(record)
break
except Exception:
continue
if not act:
return
self_ref = self
class UIR(dynamic_proxy(Runnable)):
def __init__(self, fn):
super().__init__()
self.fn = fn
def run(self):
try:
self.fn()
except Exception:
pass
def build():
Dialog = jclass("android.app.Dialog")
TextView = jclass("android.widget.TextView")
LinearLayout = jclass("android.widget.LinearLayout")
WebView = jclass("android.webkit.WebView")
WebViewClient = jclass("android.webkit.WebViewClient")
WebChromeClient = jclass("android.webkit.WebChromeClient")
DownloadListener = jclass("android.webkit.DownloadListener")
GD = jclass("android.graphics.drawable.GradientDrawable")
CD = jclass("android.graphics.drawable.ColorDrawable")
Color = jclass("android.graphics.Color")
Gravity = jclass("android.view.Gravity")
LayoutParams = jclass("android.view.WindowManager$LayoutParams")
try:
R_style = jclass("android.R$style")
field = R_style.getField("Theme_Black_NoTitleBar_Fullscreen")
theme_id = int(field.getInt(None))
dialog = Dialog(act, theme_id)
except Exception:
dialog = Dialog(act)
w = dialog.getWindow()
try:
w.setBackgroundDrawable(CD(Color.TRANSPARENT))
except Exception:
pass
try:
w.getDecorView().setPadding(0, 0, 0, 0)
except Exception:
pass
try:
flags = (
LayoutParams.FLAG_LAYOUT_NO_LIMITS |
LayoutParams.FLAG_LAYOUT_IN_SCREEN |
LayoutParams.FLAG_FULLSCREEN |
LayoutParams.FLAG_DRAWS_SYSTEM_BAR_BACKGROUNDS
)
w.setFlags(flags, flags)
except Exception:
pass
try:
lp = w.getAttributes()
lp.layoutInDisplayCutoutMode = 1
w.setAttributes(lp)
except Exception:
pass
try:
View = jclass("android.view.View")
ui = (
View.SYSTEM_UI_FLAG_LAYOUT_STABLE |
View.SYSTEM_UI_FLAG_LAYOUT_FULLSCREEN |
View.SYSTEM_UI_FLAG_LAYOUT_HIDE_NAVIGATION |
View.SYSTEM_UI_FLAG_FULLSCREEN |
View.SYSTEM_UI_FLAG_HIDE_NAVIGATION |
View.SYSTEM_UI_FLAG_IMMERSIVE_STICKY
)
w.getDecorView().setSystemUiVisibility(ui)
except Exception:
pass
dialog.setCanceledOnTouchOutside(False)
root = LinearLayout(act)
root.setOrientation(LinearLayout.VERTICAL)
try:
root.setBackgroundColor(Color.WHITE)
except Exception:
pass
root.setPadding(0, 0, 0, 0)
try:
root.setFitsSystemWindows(False)
except Exception:
pass
title = TextView(act)
title.setText(title_text)
title.setTextSize(16)
title.setTextColor(Color.BLACK)
title.setGravity(Gravity.CENTER)
title.setPadding(0, 40, 0, 16)
root.addView(title)
try:
web = WebView(act)
except Exception as e:
return
web.setLayoutParams(LinearLayout.LayoutParams(-1, 0, 1.0))
try:
s = web.getSettings()
s.setJavaScriptEnabled(True)
s.setDomStorageEnabled(True)
s.setAllowFileAccess(True)
s.setLoadWithOverviewMode(True)
s.setUseWideViewPort(True)
s.setJavaScriptCanOpenWindowsAutomatically(False)
s.setSupportMultipleWindows(False)
if UA:
s.setUserAgentString(UA)
try:
s.setMixedContentMode(0)
except Exception:
pass
except Exception:
pass
try:
JSBridge = jclass("android.webkit.JavascriptInterface")
class Bridge(dynamic_proxy(JSBridge)):
def closeDialog(self):
try:
dialog.dismiss()
except Exception:
pass
return "ok"
web.addJavascriptInterface(Bridge(), "TB")
except Exception:
pass
try:
class ClientCls(dynamic_proxy(WebViewClient)):
def shouldOverrideUrlLoading(self, view, u):
try:
s_url = str(u.toString())
if not (s_url.startswith("http://") or
s_url.startswith("https://")):
return True
if ALLOW_DOMAIN and not s_url.startswith(ALLOW_DOMAIN):
return True
view.loadUrl(s_url)
return True
except Exception:
return True
def onReceivedSslError(self, view, handler, error):
handler.proceed()
web.setWebViewClient(ClientCls())
except Exception:
pass
try:
class ChromeCls(dynamic_proxy(WebChromeClient)):
def onCreateWindow(self, view, isDialog, isUserGesture, resultMsg):
return False
web.setWebChromeClient(ChromeCls())
except Exception:
pass
try:
class DL(dynamic_proxy(DownloadListener)):
def onDownloadStart(self, u, ua, cd, mime, length):
return
web.setDownloadListener(DL())
except Exception:
pass
try:
web.loadUrl(url)
except Exception:
pass
root.addView(web)
dialog.setContentView(root)
try:
class DismissListener(dynamic_proxy(
jclass("android.content.DialogInterface$OnDismissListener")
)):
def onDismiss(self, dlg):
self_ref._read_clip_and_push()
dialog.setOnDismissListener(DismissListener())
except Exception:
pass
try:
w.setGravity(Gravity.TOP | Gravity.START)
except Exception:
pass
try:
w.setLayout(LayoutParams.MATCH_PARENT, LayoutParams.MATCH_PARENT)
except Exception:
try:
w.setLayout(-1, -1)
except Exception:
pass
dialog.show()
try:
View = jclass("android.view.View")
ui = (
View.SYSTEM_UI_FLAG_LAYOUT_STABLE |
View.SYSTEM_UI_FLAG_LAYOUT_FULLSCREEN |
View.SYSTEM_UI_FLAG_LAYOUT_HIDE_NAVIGATION |
View.SYSTEM_UI_FLAG_FULLSCREEN |
View.SYSTEM_UI_FLAG_HIDE_NAVIGATION |
View.SYSTEM_UI_FLAG_IMMERSIVE_STICKY
)
w.getDecorView().setSystemUiVisibility(ui)
except Exception:
pass
try:
w.setLayout(LayoutParams.MATCH_PARENT, LayoutParams.MATCH_PARENT)
except Exception:
pass
act.runOnUiThread(UIR(build))
except Exception as e:
pass
threading.Thread(target=launch).start()
def _get_local_ip(self):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
finally:
s.close()
if ip and not ip.startswith("127."):
return ip
except Exception:
pass
try:
ip = socket.gethostbyname(socket.gethostname())
if ip and not ip.startswith("127."):
return ip
except Exception:
pass
return "127.0.0.1"
def _get_ip_prefix(self):
ip = self._get_local_ip()
if not ip or ip.startswith("127."):
return ""
parts = ip.split(".")
if len(parts) != 4:
return ""
return ".".join(parts[:3]) + "."
def _sanitize_filename(self, name):
name = str(name or "file")
for ch in '\\/:*?"<>|':
name = name.replace(ch, '_')
name = name.strip() or "file"
if len(name) > 200:
name = name[:200]
return name
def _format_size(self, size):
try:
size = int(size)
except Exception:
return "?"
if size < 1024:
return "{} B".format(size)
elif size < 1024 * 1024:
return "{:.1f} KB".format(size / 1024)
elif size < 1024 * 1024 * 1024:
return "{:.1f} MB".format(size / 1024 / 1024)
return "{:.2f} GB".format(size / 1024 / 1024 / 1024)
def _normalize_addr(self, user_input, default_port=SHARE_PORT_START):
s = str(user_input or "").strip()
if not s:
return ""
if s.startswith(SHARE_FLAG):
s = s[len(SHARE_FLAG):].strip()
if s.startswith(("http://", "https://")):
return s.rstrip("/")
s = s.replace("http://", "").replace("https://", "").strip("/")
if re.match(r"^\d{1,3}(?:\.\d{1,3}){3}", s):
if ":" not in s:
s += ":" + str(default_port)
return "http://" + s
prefix = self._get_ip_prefix()
if prefix:
m = re.match(r"^(\d{1,3})(?::(\d{1,5}))?$", s)
if m:
last = m.group(1)
port = m.group(2) or str(default_port)
return "http://{}{}:{}".format(prefix, last, port)
return "http://" + s
def _copy_to_clipboard(self, text):
try:
from java import jclass
ctx = jclass("android.app.ActivityThread").currentApplication()
cm = ctx.getSystemService(ctx.CLIPBOARD_SERVICE)
ClipData = jclass("android.content.ClipData")
cm.setPrimaryClip(ClipData.newPlainText("", text))
return True
except Exception:
return False
def _list_received_files(self):
results = []
if not os.path.isdir(RECV_FOLDER):
return results
try:
for name in os.listdir(RECV_FOLDER):
if name.startswith("."):
continue
full = os.path.join(RECV_FOLDER, name)
if not os.path.isfile(full):
continue
try:
stat = os.stat(full)
results.append({"name": name, "path": full,
"size": stat.st_size, "mtime": stat.st_mtime})
except Exception:
continue
except Exception:
pass
results.sort(key=lambda x: x["mtime"], reverse=True)
return results
def _guess_mime(self, filename):
ext = os.path.splitext(str(filename))[1].lower()
mimes = {
".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png",
".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp",
".mp4": "video/mp4", ".mkv": "video/x-matroska", ".avi": "video/x-msvideo",
".mov": "video/quicktime", ".flv": "video/x-flv", ".ts": "video/mp2t",
".mp3": "audio/mpeg", ".wav": "audio/wav", ".flac": "audio/flac",
".m4a": "audio/mp4", ".ogg": "audio/ogg",
".txt": "text/plain", ".json": "application/json",
".pdf": "application/pdf", ".zip": "application/zip",
".py": "text/x-python", ".apk": "application/vnd.android.package-archive",
".html": "text/html", ".htm": "text/html",
}
return mimes.get(ext, "*/*")
def _open_received_file(self, activity, item):
try:
from java import jclass
toast_class = jclass("android.widget.Toast")
path = item.get("path", "")
if not path or not os.path.isfile(path):
try:
toast_class.makeText(activity, "文件不存在", toast_class.LENGTH_SHORT).show()
except Exception:
pass
return
try:
StrictMode = jclass("android.os.StrictMode")
VmPolicyBuilder = jclass("android.os.StrictMode$VmPolicy$Builder")
StrictMode.setVmPolicy(VmPolicyBuilder().build())
except Exception:
pass
Intent = jclass("android.content.Intent")
Uri = jclass("android.net.Uri")
File = jclass("java.io.File")
mime = self._guess_mime(path)
try:
intent = Intent(Intent.ACTION_VIEW)
intent.setDataAndType(Uri.fromFile(File(path)), mime)
intent.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
try:
intent.addFlags(Intent.FLAG_GRANT_READ_URI_PERMISSION)
except Exception:
pass
activity.startActivity(intent)
return
except Exception:
pass
try:
intent2 = Intent(Intent.ACTION_VIEW)
intent2.setDataAndType(Uri.fromFile(File(path)), "*/*")
intent2.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
activity.startActivity(intent2)
return
except Exception:
pass
try:
toast_class.makeText(activity, "无法打开\n" + path, toast_class.LENGTH_LONG).show()
except Exception:
pass
except Exception:
pass
def _stop_send_server(self):
if self._send_server is not None:
try:
self._send_server.shutdown()
self._send_server.server_close()
except Exception:
pass
self._send_server = None
self._send_port = 0
self._send_file_path = ""
self._send_file_name = ""
def _build_send_handler(self):
owner = self
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
try:
self._handle()
except Exception as e:
try:
self._send_simple(500, "text/plain; charset=utf-8",
"Error: {}".format(e).encode("utf-8"))
except Exception:
pass
def _handle(self):
path = urllib.parse.unquote(
urllib.parse.urlparse(self.path).path.strip("/")
)
if path == "info":
try:
size = os.path.getsize(owner._send_file_path)
except Exception:
size = 0
payload = json.dumps({
"py_helper_file": True,
"name": owner._send_file_name,
"size": size,
"version": owner.VERSION,
}, ensure_ascii=False).encode("utf-8")
self._send_simple(200, "application/json; charset=utf-8", payload)
return
if path in ("", "index.html"):
self._send_index()
return
if path == "download":
fp = owner._send_file_path
if not os.path.isfile(fp):
self._send_simple(404, "text/plain; charset=utf-8", b"File not found")
return
try:
size = os.path.getsize(fp)
except Exception:
size = 0
fn = owner._send_file_name
encoded = urllib.parse.quote(fn)
self.send_response(200)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Length", str(size))
self.send_header(
"Content-Disposition",
'attachment; filename="{}"; filename*=UTF-8\'\'{}'.format(encoded, encoded)
)
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
with open(fp, "rb") as f:
while True:
chunk = f.read(FILE_CHUNK)
if not chunk:
break
self.wfile.write(chunk)
return
self._send_simple(404, "text/plain; charset=utf-8", b"Not Found")
def _send_simple(self, code, ctype, data):
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(data)
def _send_index(self):
ip = owner._get_local_ip()
port = owner._send_port
base = "http://{}:{}".format(ip, port)
try:
size_str = owner._format_size(os.path.getsize(owner._send_file_path))
except Exception:
size_str = "?"
name = owner._send_file_name
html = (
''
''
'PY助手 · 发送'
''
'
📤 PY助手 · 发送
'
'
' + base + '
'
'
'
'
📄
'
'
' + name + '
'
'
' + size_str + '
'
'
下载'
'