#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ph-watch：连麦看 P 站的小工具。把 P 站的片子投到同一个 Wi-Fi 里的电视/投影仪上，
音量先调好再放，还能截最近 50 秒的 6 格拼图给你的 AI 看画面。全程只在你自己的电脑上跑。

  python3 ph-watch.py devices                        看看局域网里能投哪些电视/投影仪
  python3 ph-watch.py search 关键词                   搜片：viewkey  时长  标题
  python3 ph-watch.py play <链接|viewkey> [--vol N]   投上去放；写了音量就先调好再放，调不上就不放
  python3 ph-watch.py play <链接|viewkey> --dry       只试不投：取地址、起本机中转、拉第一段
  python3 ph-watch.py snap [--at 秒]                  截最近 50 秒（或指定位置往前 50 秒）的 6 格拼图，打印图片路径
  python3 ph-watch.py state | pause | resume | vol N
  python3 ph-watch.py stop                           停止播放，顺手关掉本机中转

环境变量（都可以不设）：
  PHW_PROXY   http 代理，比如 http://127.0.0.1:7890；不设就走系统代理/系统网络
  PHW_TV      电视名字里的几个字或它的 IP，局域网里有好几台时用来指定
  PHW_PORT    本机中转用的端口，默认 8766
  PHW_HEIGHT  画质上限，默认 720
"""
import html, importlib.util, ipaddress, json, os, re, shutil, socket, subprocess, sys, threading, time
import urllib.error, urllib.parse, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from xml.etree import ElementTree as ET

HOME = os.path.join(os.path.expanduser("~"), ".ph-watch")
PORT = int(os.environ.get("PHW_PORT", "8766"))
HEIGHT = int(os.environ.get("PHW_HEIGHT", "720"))
PROXY = os.environ.get("PHW_PROXY", "").strip()
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0 Safari/537.36"
SITE = "https://www.pornhub.com"
REF = {"Referer": SITE + "/", "Origin": SITE}  # P 站的分片不带这两个头会回 404
NET = urllib.request.build_opener(urllib.request.ProxyHandler({"http": PROXY, "https": PROXY}) if PROXY else urllib.request.ProxyHandler())
LAN = urllib.request.build_opener(urllib.request.ProxyHandler({}))  # 本机和电视之间不走代理
SVC = {"avt": "urn:schemas-upnp-org:service:AVTransport:1", "rc": "urn:schemas-upnp-org:service:RenderingControl:1"}


def say(*a):
    print(*a, flush=True)


def die(msg):
    say(msg)
    sys.exit(1)


def path(*p):
    os.makedirs(HOME, exist_ok=True)
    return os.path.join(HOME, *p)


def jload(name):
    try:
        with open(path(name), encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def jsave(name, obj):
    with open(path(name), "w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False)


def fetch(url, headers=None, opener=NET, tries=4, timeout=30):
    err = None
    for i in range(tries):
        try:
            return opener.open(urllib.request.Request(url, headers=headers or {"User-Agent": UA}), timeout=timeout).read()
        except urllib.error.HTTPError as e:
            if e.code in (403, 404, 410):
                raise
            err = e
        except Exception as e:  # 隔着梯子偶尔断一下（TLS bad record mac 之类），重来就好
            err = e
        time.sleep(0.5 * (i + 1))
    raise err


def hms(n):
    return "%02d:%02d:%02d" % (n // 3600, n % 3600 // 60, n % 60)


def secs(t):
    try:
        h, m, s = (t or "0:0:0").split(".")[0].split(":")
        return int(h) * 3600 + int(m) * 60 + int(s)
    except Exception:
        return 0


def clear_snaps():
    for f in os.listdir(path()):
        if f.startswith("snap-") and f.endswith(".jpg"):
            try:
                os.remove(path(f))
            except OSError:
                pass


# ───────── P 站：取地址、搜片 ─────────

def key_of(s):
    m = re.search(r"viewkey=([0-9a-z]+)", s)
    if m:
        return m.group(1)
    if re.fullmatch(r"[0-9a-z]+", s):
        return s
    die("认不出这个链接：要 P 站 view_video.php?viewkey=… 的链接，或者 viewkey 本身")


def reachable():
    try:
        fetch(SITE + "/", tries=1, timeout=12)
        return True
    except Exception:
        return False


def ytdlp():
    exe = shutil.which("yt-dlp")
    if exe:
        return [exe]
    if importlib.util.find_spec("yt_dlp"):
        return [sys.executable, "-m", "yt_dlp"]
    die("没找到 yt-dlp：先 pip install -U yt-dlp")


def resolve(key):
    cmd = ytdlp() + ["-j", "--no-warnings", "--add-header", "Referer:" + SITE + "/"] + (["--proxy", PROXY] if PROXY else [])
    err = "试了 5 次都只拿到整段 mp4，没有分段流"
    for _ in range(5):  # 隔着梯子取地址偶尔空手，多试几次
        r = subprocess.run(cmd + [SITE + "/view_video.php?viewkey=" + key], capture_output=True, text=True, timeout=120)
        if r.returncode == 0:
            d = json.loads(r.stdout)
            hls = [f for f in d.get("formats", []) if str(f.get("protocol", "")).startswith("m3u8")]
            if hls:
                break
        else:
            err = next((l for l in reversed(r.stderr.splitlines()) if l.startswith("ERROR")), err)
        time.sleep(2)
    else:
        die("取不到这部的地址：" + err[:200])
    ok = [f for f in hls if (f.get("height") or 0) <= HEIGHT]
    f = max(ok, key=lambda f: (f.get("height") or 0, f.get("tbr") or 0)) if ok else min(hls, key=lambda f: f.get("height") or 0)
    m = {"key": key, "url": f["url"], "headers": f.get("http_headers") or {"User-Agent": UA}, "title": d.get("title") or key,
         "duration": int(d.get("duration") or 0), "height": f.get("height"), "at": int(time.time())}
    jsave("v-" + key + ".json", m)
    return m


def search(q):
    if not reachable():
        die("打不开 P 站：梯子开了吗？命令行走不走梯子，看说明里的 PHW_PROXY")
    t = fetch(SITE + "/video/search?search=" + urllib.parse.quote_plus(q)).decode("utf-8", "ignore")
    seen, n = set(), 0
    for k in re.finditer(r'data-video-vkey="([0-9a-z]+)"', t):
        if k.group(1) in seen:
            continue
        blk = t[k.start():k.start() + 5000]
        ti = re.search(r'href="/view_video\.php\?viewkey=[0-9a-z]+"[^>]*title="([^"]+)"', blk)
        du = re.search(r'class="[^"]*\bduration\b[^"]*"[^>]*>\s*([\d:]+)', blk)
        if not ti:
            continue
        seen.add(k.group(1))
        say(k.group(1), du.group(1) if du else "?", html.unescape(ti.group(1))[:90])
        n += 1
        if n >= 12:
            break
    if not n:
        say("没搜到，换个词试试")


# ───────── 本机中转：电视 → 这里 → 梯子 → P 站 ─────────
# 电视自己多半够不着 P 站，CDN 还要 Referer、只认取地址时的那个出口，所以由电脑代取。只认登记过的片子，不是开放代理。

SEGS, CACHE, ORDER, BUSY, LASTREF = {}, {}, [], set(), {}
LOCK, RLOCK = threading.Lock(), threading.Lock()


def vmeta(key):
    m = jload("v-" + key + ".json")
    if not m:
        raise KeyError(key)
    return m


def vheaders(m):
    return dict(m["headers"], **REF)


def load(key, refresh=False):
    if refresh:
        with RLOCK:
            if time.time() - LASTREF.get(key, 0) > 60:  # 地址两小时左右过期，过期就重取
                subprocess.run([sys.executable, os.path.abspath(__file__), "refresh", key], capture_output=True, timeout=650)
                LASTREF[key] = time.time()
    m = vmeta(key)
    pl = fetch(m["url"], vheaders(m)).decode()
    base = m["url"].split("?")[0].rsplit("/", 1)[0] + "/"
    segs, out = [], []
    for l in pl.splitlines():
        if l and not l.startswith("#"):
            out.append("%d.ts" % len(segs))
            segs.append(l if l.startswith("http") else base + l)
        else:
            out.append(l)
    with LOCK:
        SEGS[key] = segs
    return "\n".join(out) + "\n"


def seg(key, n):
    with LOCK:
        b = CACHE.get((key, n))
    if b is not None:
        return b
    if key not in SEGS:
        load(key)
    try:
        b = fetch(SEGS[key][n], vheaders(vmeta(key)))
    except urllib.error.HTTPError:
        load(key, refresh=True)
        b = fetch(SEGS[key][n], vheaders(vmeta(key)))
    with LOCK:
        CACHE[(key, n)] = b
        ORDER.append((key, n))
        while len(ORDER) > 40:  # 留最近几分钟的片段，截图不用再隔着梯子重下
            CACHE.pop(ORDER.pop(0), None)
    return b


def prefetch(key, n):
    for i in (n + 1, n + 2):
        k = (key, i)
        with LOCK:
            if i >= len(SEGS.get(key, [])) or k in CACHE or k in BUSY:
                continue
            BUSY.add(k)
        try:
            seg(key, i)
        except Exception as e:
            say("prefetch", key, i, repr(e)[:120])
        finally:
            with LOCK:
                BUSY.discard(k)


class Relay(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def reply(self, code, body, ctype):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)

    def do_HEAD(self):
        self.do_GET()

    def do_GET(self):
        p = self.path.split("?")[0]
        if not ipaddress.ip_address(self.client_address[0]).is_private:  # 只给家里局域网用，外网来的一律拒
            return self.reply(403, b"no", "text/plain")
        if p == "/quit" and self.client_address[0] == "127.0.0.1":  # 只有本机能关掉中转
            self.reply(200, b"bye", "text/plain")
            return threading.Thread(target=self.server.shutdown, daemon=True).start()
        try:
            if p == "/ping":
                return self.reply(200, b"ok", "text/plain")
            m = re.fullmatch(r"/h/([0-9a-z]+)/index\.m3u8", p)
            if m:
                return self.reply(200, load(m.group(1)).encode(), "application/vnd.apple.mpegurl")
            m = re.fullmatch(r"/h/([0-9a-z]+)/(\d+)\.ts", p)
            if m:
                key, n = m.group(1), int(m.group(2))
                for _ in range(300):  # 预取正在拿这一段就等它，别重复下
                    with LOCK:
                        if (key, n) not in BUSY:
                            break
                    time.sleep(0.2)
                b = seg(key, n)
                threading.Thread(target=prefetch, args=(key, n), daemon=True).start()
                return self.reply(200, b, "video/mp2t")
            self.reply(404, b"no", "text/plain")
        except Exception as e:
            say("ERR", p, repr(e)[:200])
            try:
                self.reply(502, b"upstream", "text/plain")
            except Exception:
                pass


def serve():
    ThreadingHTTPServer.daemon_threads = True
    srv = ThreadingHTTPServer(("0.0.0.0", PORT), Relay)
    say(time.strftime("%m-%d %H:%M:%S"), "中转开在 %d 端口" % PORT)
    srv.serve_forever()


def relay_up():
    try:
        return fetch("http://127.0.0.1:%d/ping" % PORT, opener=LAN, tries=1, timeout=3) == b"ok"
    except Exception:
        return False


def quit_relay():
    try:
        fetch("http://127.0.0.1:%d/quit" % PORT, opener=LAN, tries=1, timeout=3)
    except Exception:
        pass


def ensure_relay():
    if relay_up():
        return
    kw = {"creationflags": 0x00000008 | 0x00000200} if os.name == "nt" else {"start_new_session": True}
    subprocess.Popen([sys.executable, os.path.abspath(__file__), "serve"], stdout=open(path("relay.log"), "a"),
                     stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, **kw)
    for _ in range(30):
        time.sleep(0.3)
        if relay_up():
            return
    die("本机中转没起来（端口 %d 被占了？换个 PHW_PORT 试试），细节看 %s" % (PORT, path("relay.log")))


# ───────── 电视/投影仪：DLNA 遥控 ─────────

def describe(loc):
    root = ET.fromstring(fetch(loc, opener=LAN, tries=2, timeout=5))
    for el in root.iter():
        if isinstance(el.tag, str) and "}" in el.tag:
            el.tag = el.tag.split("}", 1)[1]
    base = (root.findtext("URLBase") or loc).strip()
    d = {"name": (root.findtext(".//device/friendlyName") or "?").strip(), "ip": urllib.parse.urlparse(loc).hostname, "loc": loc}
    for sv in root.iter("service"):
        t, c = sv.findtext("serviceType") or "", (sv.findtext("controlURL") or "").strip()
        if c and ":AVTransport:" in t:
            d["avt"] = urllib.parse.urljoin(base, c)
        if c and ":RenderingControl:" in t:
            d["rc"] = urllib.parse.urljoin(base, c)
    return d


def discover(wait=3):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
    s.settimeout(wait)
    msg = ('M-SEARCH * HTTP/1.1\r\nHOST: 239.255.255.250:1900\r\nMAN: "ssdp:discover"\r\nMX: 2\r\n'
           "ST: urn:schemas-upnp-org:device:MediaRenderer:1\r\n\r\n").encode()
    for _ in range(2):
        s.sendto(msg, ("239.255.255.250", 1900))
    locs, end = [], time.time() + wait
    while time.time() < end:
        try:
            data = s.recvfrom(65535)[0].decode("utf-8", "ignore")
        except socket.timeout:
            break
        m = re.search(r"(?im)^location:\s*(\S+)", data)
        if m and m.group(1) not in locs:
            locs.append(m.group(1))
    s.close()
    devs = []
    for loc in locs:
        try:
            d = describe(loc)
            if d.get("avt"):
                devs.append(d)
        except Exception:
            pass
    return devs


def soap(d, svc, action, args=""):
    if not d.get(svc):
        raise RuntimeError("这台设备没有 " + svc)
    body = ('<?xml version="1.0" encoding="utf-8"?><s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
            's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:%s xmlns:u="%s"><InstanceID>0</InstanceID>%s</u:%s>'
            "</s:Body></s:Envelope>" % (action, SVC[svc], args, action)).encode()
    r = urllib.request.Request(d[svc], data=body, headers={"Content-Type": 'text/xml; charset="utf-8"', "SOAPAction": '"%s#%s"' % (SVC[svc], action)})
    return LAN.open(r, timeout=8).read().decode("utf-8", "ignore")


def tv():
    want = os.environ.get("PHW_TV", "").strip()
    c = jload("tv.json")
    if c and (not want or want in c["name"] or want == c["ip"]):
        try:
            soap(c, "avt", "GetTransportInfo")
            return c
        except Exception:
            pass  # 电视重启以后端口常会变，重新找
    devs = discover()
    if want:
        devs = [d for d in devs if want in d["name"] or want == d["ip"]]
    for d in devs:
        try:
            soap(d, "avt", "GetTransportInfo")
            jsave("tv.json", d)
            return d
        except Exception:
            continue
    die("局域网里没找到能投的电视/投影仪：电脑和电视连的是同一个 Wi-Fi 吗？电视开着吗？" + ("（PHW_TV=%s 没对上）" % want if want else ""))


def state(d):
    ti, pi = soap(d, "avt", "GetTransportInfo"), soap(d, "avt", "GetPositionInfo")

    def g(tag, s):
        m = re.search(r"<%s>(.*?)</%s>" % (tag, tag), s)
        return m.group(1) if m else ""
    return g("CurrentTransportState", ti), g("RelTime", pi), g("TrackDuration", pi)


def setvol(d, v):
    for _ in range(3):  # 有的电视一下没应，多按两下
        try:
            soap(d, "rc", "SetVolume", "<Channel>Master</Channel><DesiredVolume>%d</DesiredVolume>" % v)
            return True
        except Exception:
            time.sleep(1)
    return False


def lan_ip_for(d):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect((d["ip"], 1900))
        return s.getsockname()[0]
    finally:
        s.close()


# ───────── 命令 ─────────

def play(arg, vol, dry):
    key = key_of(arg)
    d = None if dry else tv()
    if vol is not None and d and not setvol(d, vol):  # 先调音量再取片：中途被打断，声音也已经是要的大小
        die("电视没接住音量，先不放（免得一开声就很大）")
    if not reachable():
        die("打不开 P 站：梯子开了吗？命令行走不走梯子，看说明里的 PHW_PROXY")
    m = resolve(key)
    ensure_relay()
    local = "http://127.0.0.1:%d/h/%s/" % (PORT, key)
    n = sum(1 for l in fetch(local + "index.m3u8", opener=LAN, timeout=60).decode().splitlines() if l and not l.startswith("#"))
    first = fetch(local + "0.ts", opener=LAN, timeout=120)
    say("片子：%s（%s，%sp，%d 段，第一段 %.1fMB 已到本机中转）" % (m["title"], hms(m["duration"]), m["height"], n, len(first) / 1e6))
    jsave("last.json", {"key": key, "title": m["title"], "duration": m["duration"], "dry": dry})
    clear_snaps()
    if dry:
        say("只试不投：本机中转地址 %sindex.m3u8，可以用 VLC 打开看看" % local)
        return
    url = "http://%s:%d/h/%s/index.m3u8" % (lan_ip_for(d), PORT, key)
    meta = ('<DIDL-Lite xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/" xmlns:dc="http://purl.org/dc/elements/1.1/" '
            'xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"><item id="1" parentID="0" restricted="1"><dc:title>video</dc:title>'
            '<upnp:class>object.item.videoItem</upnp:class><res protocolInfo="http-get:*:application/vnd.apple.mpegurl:*">%s</res>'
            "</item></DIDL-Lite>" % html.escape(url))
    soap(d, "avt", "SetAVTransportURI", "<CurrentURI>%s</CurrentURI><CurrentURIMetaData>%s</CurrentURIMetaData>" % (html.escape(url), html.escape(meta)))
    soap(d, "avt", "Play", "<Speed>1</Speed>")
    st = pos = dur = ""
    for _ in range(20):
        time.sleep(1)
        st, pos, dur = state(d)
        if st == "PLAYING":
            break
    if vol is not None:
        setvol(d, vol)  # 有的电视换片会把音量带回去，放起来再按一次
    say("%s：%s %s / %s%s" % (d["name"], st, pos, dur, "，音量 %d" % vol if vol is not None else ""))


def snap(at):
    last = jload("last.json")
    if not last:
        die("还没用 play 放过片子")
    if at is None:
        if last.get("dry"):
            die("上一次只是试（--dry），没投到电视上；要截就用 --at 秒 指定位置")
        st, pos, _ = state(tv())
        if st not in ("PLAYING", "PAUSED_PLAYBACK"):
            die("电视上没在放（%s）" % (st or "没回应"))
        at = secs(pos)
    ff = shutil.which("ffmpeg") or die("没找到 ffmpeg，先装上")
    s0, span = max(0, at - 50), min(at, 50) + 1
    clear_snaps()
    out = path("snap-%s.jpg" % time.strftime("%H%M%S"))
    ensure_relay()
    r = subprocess.run([ff, "-v", "error", "-y", "-skip_frame", "nokey", "-ss", str(s0), "-i", "http://127.0.0.1:%d/h/%s/index.m3u8" % (PORT, last["key"]),
                        "-t", str(span), "-an", "-sn", "-dn", "-vf", "fps=1/10,scale=640:360,tile=3x2", "-frames:v", "1", out],
                       capture_output=True, text=True, timeout=180)
    if r.returncode != 0 or not os.path.exists(out):
        die("截图没成功：" + ((r.stderr.strip().splitlines() or ["?"])[-1])[:200])
    say("%s\n截的是 %s～%s：6 格每 10 秒一帧，左→右、上→下由早到晚，右下角最新；重放或换片以后这张就作废了，再截一次。" % (out, hms(s0), hms(at)))


def main():
    a = sys.argv[1:]
    if not a:
        return say(__doc__)
    c = a[0]

    def opt(name):
        return a[a.index(name) + 1] if name in a and a.index(name) + 1 < len(a) else None
    if c == "serve":
        serve()
    elif c == "devices":
        devs = discover()
        for d in devs:
            say("%s  %s%s" % (d["name"], d["ip"], "" if d.get("rc") else "（不能调音量）"))
        if not devs:
            say("没找到：电脑和电视连的是同一个 Wi-Fi 吗？")
    elif c == "search" and len(a) > 1:
        search(" ".join(a[1:]))
    elif c == "play" and len(a) > 1:
        v = opt("--vol")
        play(a[1], int(v) if v and v.isdigit() else None, "--dry" in a)
    elif c == "snap":
        v = opt("--at")
        snap(int(v) if v and v.isdigit() else None)
    elif c == "refresh" and len(a) > 1:
        resolve(key_of(a[1]))
    elif c == "vol" and len(a) > 1 and a[1].isdigit():
        say("音量 " + a[1] if setvol(tv(), int(a[1])) else "电视没接住音量")
    elif c in ("pause", "resume", "stop", "state"):
        d = tv()
        if c == "pause":
            soap(d, "avt", "Pause")
        elif c == "resume":
            soap(d, "avt", "Play", "<Speed>1</Speed>")
        elif c == "stop":
            soap(d, "avt", "Stop")
            clear_snaps()
            quit_relay()
        st, pos, dur = state(d)
        say("%s：%s %s / %s  %s" % (d["name"], st, pos, dur, ((jload("last.json") or {}).get("title") or "")[:60]))
    else:
        say(__doc__)


if __name__ == "__main__":
    main()