#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
DeepSeek 中转 v3 - Windows 客户端（零配置，纯中转路线 A）
安装即用：双击运行即自动接入互通，无需填写任何 token / Key / 服务器地址。

架构（路线 A）：
  收任务 → 写入 workdir/tasks/ （harness 监视此目录自动处理）
  harness 处理 → 产物落到 workdir/output/ → 客户端自动上传服务器
  服务器广播 → 所有机器自动下载产物到本机 output/

命令行：
  client.py                          # 常驻 + 交互
  client.py --headless               # 后台常驻（无交互）
  client.py task --target win-01 内容 # 发指定目标任务
  client.py task --target all 内容   # 广播
  client.py status                   # 查看状态

打包 exe（任意 Windows 机器跑一次）：
  pip install pyinstaller
  pyinstaller --onefile --name DeepSeekSync client.py
"""
import argparse
import base64
import hashlib
import json
import os
import queue as _queue
import socket
import struct
import sys
import threading
import time
import urllib.request
import urllib.parse
import uuid
from pathlib import Path

# ============ 内置配置（零配置，一般无需改动） ============
SERVER = "82.156.163.155"
PORT = 8282
AUTH = "ds-relay-2k26-local"
TASK_FILE = "client.py"   # 自己的文件名，用于定位同目录 config.json
# 允许用同目录 config.json 覆盖（可选，不写也完全可用）
# ----------------------------------------------------------

CFG_OVERRIDE = {}
try:
    _cfg_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                             "config.json")
    if os.path.isfile(_cfg_path):
        with open(_cfg_path, "r", encoding="utf-8") as _f:
            CFG_OVERRIDE = json.load(_f)
except (IOError, ValueError):
    pass

DEVICE = CFG_OVERRIDE.get("device", socket.gethostname())
WORKDIR = CFG_OVERRIDE.get("workdir") or \
    os.path.join(os.path.expanduser("~"), "deepseek-relay")
TASKS_DIR = os.path.join(WORKDIR, "tasks")
OUTPUT_DIR = os.path.join(WORKDIR, "output")
for _d in (TASKS_DIR, OUTPUT_DIR, WORKDIR):
    try:
        os.makedirs(_d, exist_ok=True)
    except OSError:
        pass

# 已下载文件（避免重复）
LOCAL_FILES = {}          # fid -> {"name", "path"}
LOCAL_FILES_LOCK = threading.Lock()
STOP = threading.Event()

WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
DEFAULT_STABLE_SEC = 1.0  # 输出文件大小稳定多久才算写完


# ============ 工具 ============
def ws_accept(key):
    return base64.b64encode(hashlib.sha1(
        (key + WS_MAGIC).encode("utf-8")).digest()).decode()


def recv_exact(sock, n):
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("connection closed")
        buf += chunk
    return buf


def recv_frame(sock):
    h = recv_exact(sock, 2)
    opcode = h[0] & 0x0F
    ln = h[1] & 0x7F
    masked = h[1] >> 7
    if ln == 126:
        ln = struct.unpack(">H", recv_exact(sock, 2))[0]
    elif ln == 127:
        ln = struct.unpack(">Q", recv_exact(sock, 8))[0]
    if masked:
        mask = recv_exact(sock, 4)
        data = bytearray(recv_exact(sock, ln))
        for i in range(ln):
            data[i] ^= mask[i % 4]
        return opcode, bytes(data)
    return opcode, recv_exact(sock, ln)


def send_frame(sock, opcode, data):
    hdr = bytes([0x80 | opcode])
    n = len(data)
    if n < 126:
        hdr += bytes([n])
    elif n < 65536:
        hdr += bytes([126]) + struct.pack(">H", n)
    else:
        hdr += bytes([127]) + struct.pack(">Q", n)
    mask = os.urandom(4)
    body = bytes(b ^ mask[i % 4] for i, b in enumerate(data))
    sock.sendall(hdr + bytes([0x80]) + mask + body)


def send_json(sock, obj):
    send_frame(sock, 1, json.dumps(obj, ensure_ascii=False).encode("utf-8"))


def api_url(path, extra=""):
    sep = "&" if ("?" in path or extra) else "?"
    return "http://{}:{}{}{}auth={}{}".format(
        SERVER, PORT, path, sep, AUTH, extra)


def http_request(url, method="GET", data=None, headers=None, timeout=30):
    req = urllib.request.Request(url, data=data, method=method)
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return resp.status, resp.read()


# ============ WebSocket 连接 ============
def ws_connect():
    sock = socket.create_connection((SERVER, PORT), timeout=15)
    key = base64.b64encode(os.urandom(16)).decode()
    req = ("GET /ws?auth={}&device={} HTTP/1.1\r\n"
           "Host: {}:{}\r\n"
           "Upgrade: websocket\r\n"
           "Connection: Upgrade\r\n"
           "Sec-WebSocket-Key: {}\r\n"
           "Sec-WebSocket-Version: 13\r\n\r\n").format(
        AUTH, urllib.parse.quote(DEVICE),
        SERVER, PORT, key)
    sock.sendall(req.encode("utf-8"))
    resp = b""
    while b"\r\n\r\n" not in resp:
        chunk = sock.recv(4096)
        if not chunk:
            raise ConnectionError("handshake failed")
        resp += chunk
    if b"101" not in resp.split(b"\r\n")[0]:
        raise ConnectionError("handshake rejected: "
                              + resp[:100].decode("latin-1"))
    return sock


# ============ 任务落盘 ============
def write_task_files(task):
    tid = task.get("task_id", uuid.uuid4().hex[:12])
    prompt = task.get("prompt", "")
    target = task.get("target", "")
    from_d = task.get("from", "")
    json_path = os.path.join(TASKS_DIR, "{}.json".format(tid))
    txt_path = os.path.join(TASKS_DIR, "{}.txt".format(tid))
    try:
        meta = {k: v for k, v in task.items()}
        if "type" in meta:
            del meta["type"]
        with open(json_path, "w", encoding="utf-8") as f:
            json.dump(meta, f, ensure_ascii=False, indent=2)
        with open(txt_path, "w", encoding="utf-8") as f:
            header = "# 任务 {}\n目标: {}\n来源: {}\n\n".format(
                tid, target or "(broadcast)", from_d or "?")
            f.write(header + prompt)
        return json_path, txt_path
    except IOError as e:
        print("[客户端] 写入任务文件失败: {}".format(e))
        return None, None


# ============ 下载文件 ============
def download_file(fid, dest_path):
    url = api_url("/api/file/" + fid + "?device="
                  + urllib.parse.quote(DEVICE))
    st, body = http_request(url, timeout=120)
    if st != 200:
        return False
    # 原子写入（先写 .tmp 再 rename，避免 harness 读到半成品）
    tmp = dest_path + ".downloading"
    try:
        with open(tmp, "wb") as f:
            f.write(body)
        os.replace(tmp, dest_path)
        return True
    except IOError as e:
        print("[客户端] 下载写入失败 {}: {}".format(dest_path, e))
        return False


def sync_file_list():
    """启动时拉最近 200 个文件补齐本地"""
    try:
        st, body = http_request(api_url_q("/api/status?"), timeout=20)
        if st != 200:
            return
        data = json.loads(body.decode("utf-8"))
        for r in data.get("recent_files", []):
            fid = r["id"]
            dest = os.path.join(OUTPUT_DIR, r["name"])
            with LOCAL_FILES_LOCK:
                if fid in LOCAL_FILES:
                    continue
            if os.path.isfile(dest):
                with LOCAL_FILES_LOCK:
                    LOCAL_FILES[fid] = {"name": r["name"], "path": dest}
                continue
            if download_file(fid, dest):
                with LOCAL_FILES_LOCK:
                    LOCAL_FILES[fid] = {"name": r["name"], "path": dest}
                print("[同步] 下载: {}".format(r["name"]))
    except Exception as e:
        print("[同步] 启动同步失败: {}".format(e))


def api_url_q(path):
    sep = "&" if "?" in path else "?"
    return "http://{}:{}{}{}auth={}".format(
        SERVER, PORT, path, sep, AUTH)


# ============ output/ 监视 & 上传 ============
# 临时/半成品文件过滤
TEMP_SUFFIXES = (".downloading", ".tmp", ".part", ".crdownload",
                 ".partial", ".~")
TEMP_PREFIXES = ("~$", ".~", "_", ".")


def is_temp(name):
    low = name.lower()
    if any(low.endswith(s) for s in TEMP_SUFFIXES):
        return True
    for p in TEMP_PREFIXES:
        if low.startswith(p):
            return True
    return False


def upload_file(path, name, device=None):
    device = device or DEVICE
    url = api_url_q("/api/upload?name=" + urllib.parse.quote(name)
                    + "&device=" + urllib.parse.quote(device))
    with open(path, "rb") as f:
        data = f.read()
    try:
        st, body = http_request(url, method="POST", data=data,
                                headers={"Content-Type": "application/octet-stream"},
                                timeout=120)
        if st == 200:
            r = json.loads(body.decode("utf-8"))
            print("[上传] 完成: {} -> id={} size={}".format(name, r.get("id"), r.get("size")))
            return r.get("id")
        else:
            print("[上传] HTTP {}: {}".format(st, body[:200]))
    except Exception as e:
        print("[上传] 失败 {}: {}".format(name, e))
    return None


def watch_output_dir():
    """监视 output/ 目录：新文件大小稳定 DEFAULT_STABLE_SEC 秒后上传"""
    # 记录每个候选文件的首次出现时间 & 首次记录的大小
    pending = {}   # path -> {"mtime": t, "size": n, "first_seen": t}
    uploaded = set()  # 已经上传过（本次运行内），避免重传

    while not STOP.is_set():
        try:
            now = time.time()
            # 扫描目录
            try:
                names = os.listdir(OUTPUT_DIR)
            except OSError:
                time.sleep(1); continue
            current_paths = set()
            for name in names:
                if is_temp(name):
                    continue
                full = os.path.join(OUTPUT_DIR, name)
                if not os.path.isfile(full):
                    continue
                current_paths.add(full)
                # 跳过从服务器下载下来的文件（避免回环上传）
                with LOCAL_FILES_LOCK:
                    if any(rec.get("path") == full
                           for rec in LOCAL_FILES.values()):
                        continue
                try:
                    st = os.stat(full)
                except OSError:
                    continue
                key = full + "|" + str(st.st_size)
                if key in uploaded:
                    continue
                if full not in pending:
                    pending[full] = {"size": st.st_size, "first_seen": now}
                else:
                    # 大小没变 → 看稳定时间
                    if pending[full]["size"] == st.st_size:
                        stable = now - pending[full]["first_seen"]
                        if stable >= DEFAULT_STABLE_SEC:
                            upload_file(full, name)
                            uploaded.add(key)
                            pending.pop(full, None)
                    else:
                        # 大小变了 → 重置稳定计时
                        pending[full] = {"size": st.st_size, "first_seen": now}
            # 清理已消失的文件
            for p in list(pending.keys()):
                if p not in current_paths:
                    pending.pop(p, None)
        except Exception as e:
            print("[监视] 异常: {}".format(e))
        STOP.wait(1.0)


# ============ 发任务（走 HTTP，避免 WS 短连接挤掉长连接） ============
def send_task(target, prompt, model=None, extra=None):
    tid = uuid.uuid4().hex[:12]
    task = {"type": "task", "task_id": tid, "target": target or "all",
            "prompt": prompt}
    if model:
        task["model"] = model
    if extra:
        task["extra"] = extra
    try:
        url = "http://{}:{}/api/task?auth={}&device={}".format(
            SERVER, PORT, AUTH, urllib.parse.quote(DEVICE))
        body = json.dumps(task, ensure_ascii=False).encode("utf-8")
        st, resp = http_request(url, method="POST", data=body,
                                headers={"Content-Type": "application/json"},
                                timeout=15)
        if st == 200:
            r = json.loads(resp.decode("utf-8"))
            print("[任务] {} → {} status={}".format(
                tid, target or "all", r.get("status")))
            return r.get("task_id", tid)
        print("[任务] HTTP {}: {}".format(st, resp[:200]))
    except Exception as e:
        print("[任务] 失败: {}".format(e))
    return None


# ============ 状态查询 ============
def get_status():
    try:
        st, body = http_request(api_url_q("/api/status?"), timeout=15)
        return json.loads(body.decode("utf-8")) if st == 200 else {}
    except Exception as e:
        return {"error": str(e)}


# ============ 常驻主循环（WebSocket + 消息处理） ============
def ws_main():
    while not STOP.is_set():
        try:
            sock = ws_connect()
            sock.settimeout(35)
            print("[连接] 已接入 服务器={}:{} 设备={}".format(
                SERVER, PORT, DEVICE))
            last_ping = 0
            while not STOP.is_set():
                now = time.time()
                if now - last_ping > 30:
                    send_json(sock, {"type": "pong"})
                    last_ping = now
                try:
                    opcode, data = recv_frame(sock)
                except socket.timeout:
                    continue
                except ConnectionError:
                    raise
                if opcode == 8:
                    break
                if opcode == 9:
                    send_frame(sock, 10, data); continue
                if opcode == 10:
                    continue
                if opcode != 1:
                    continue
                try:
                    msg = json.loads(data.decode("utf-8"))
                except (ValueError, UnicodeDecodeError):
                    continue
                _handle_msg(msg)
        except ConnectionError as e:
            print("[连接] 断开: {}，3秒后重连".format(e))
        except Exception as e:
            print("[连接] 异常: {}，3秒后重连".format(e))
        STOP.wait(3)


def _handle_msg(msg):
    t = msg.get("type")
    if t == "welcome":
        online = [d for d, m in msg.get("devices", {}).items()
                  if m.get("online")]
        print("[连接] 在线: {}".format(online))
    elif t == "devices":
        online = [d for d, m in msg.get("devices", {}).items()
                  if m.get("online")]
        print("[设备] 在线变更: {}".format(online))
    elif t == "task":
        task_id = msg.get("task_id", "?")
        target = msg.get("target", "all")
        from_d = msg.get("from", "?")
        print("[DEBUG-TASK] recv target={} from={} my_DEVICE={}".format(
            target, from_d, DEVICE))
        # 只处理广播或指定给我的
        if target not in ("all", DEVICE):
            print("[任务] {} 发给 {}, 跳过".format(task_id, target))
            return
        jp, tp = write_task_files(msg)
        if jp:
            print("[任务] {} 已落盘 {} + {} → harness 应开始处理".format(
                task_id, os.path.basename(jp), os.path.basename(tp)))
    elif t == "file_list":
        # 离线补齐
        _download_missing(msg.get("files", []))
    elif t == "sync":
        # Mac 点同步 → 服务器广播 → 自动拉取全部新文件
        print("[同步] 收到同步指令（发起: {}），拉取全部文件".format(
            msg.get("trigger", "?")))
        try:
            st, body = http_request(api_url_q("/api/status?"), timeout=20)
            if st == 200:
                data = json.loads(body.decode("utf-8"))
                _download_missing(data.get("recent_files", []))
        except Exception as e:
            print("[同步] 拉取失败: {}".format(e))
    elif t == "file":
        fid = msg.get("id")
        dest = os.path.join(OUTPUT_DIR, msg.get("name", "file"))
        with LOCAL_FILES_LOCK:
            if fid in LOCAL_FILES:
                return
        if download_file(fid, dest):
            with LOCAL_FILES_LOCK:
                LOCAL_FILES[fid] = {"name": msg.get("name"), "path": dest}
            print("[同步] 新产物: {}".format(msg.get("name")))


def _download_missing(files):
    """拉取缺失文件列表到本地（幂等，重复调用安全）"""
    for r in files or []:
        fid = r.get("id")
        if not fid:
            continue
        dest = os.path.join(OUTPUT_DIR, r.get("name", "file"))
        with LOCAL_FILES_LOCK:
            if fid in LOCAL_FILES:
                continue
        if os.path.isfile(dest):
            with LOCAL_FILES_LOCK:
                LOCAL_FILES[fid] = {"name": r.get("name"), "path": dest}
            continue
        if download_file(fid, dest):
            with LOCAL_FILES_LOCK:
                LOCAL_FILES[fid] = {"name": r.get("name"), "path": dest}
            print("[同步] 下载: {}".format(r.get("name")))


# ============ 命令行 ============
def build_argparser():
    p = argparse.ArgumentParser(prog="DeepSeekSync", description="v3 零配置中转客户端")
    p.add_argument("--server", default=SERVER)
    p.add_argument("--port", type=int, default=PORT)
    p.add_argument("--auth", default=AUTH)
    p.add_argument("--device", default=DEVICE)
    p.add_argument("--workdir", default=None)
    p.add_argument("--headless", action="store_true", default=False)
    sub = p.add_subparsers(dest="cmd")
    pt = sub.add_parser("task", help="发任务")
    pt.add_argument("--target", default="all",
                    help="目标设备名，或 all 广播")
    pt.add_argument("prompt", nargs=argparse.REMAINDER)
    sub.add_parser("status", help="查看服务器状态")
    sub.add_parser("list", help="在线设备")
    return p


def interactive():
    print("DeepSeekSync v3 已启动。命令:")
    print("  task --target <机器名|all> <内容>   发任务")
    print("  list                                  在线设备")
    print("  status                                服务器状态")
    print("  quit                                  退出")
    while not STOP.is_set():
        try:
            line = sys.stdin.readline()
        except KeyboardInterrupt:
            break
        if not line:
            break
        line = line.strip()
        if not line:
            continue
        if line == "quit":
            STOP.set(); break
        if line == "status":
            print(json.dumps(get_status(), ensure_ascii=False, indent=2))
        elif line == "list":
            st = get_status()
            ds = st.get("devices", {})
            print("在线: {}".format(
                [d for d, m in ds.items() if m.get("online")]))
            print("离线队列: {}".format(st.get("queue_per_device", {})))
        elif line.startswith("task "):
            rest = line[5:]
            target = "all"
            if rest.startswith("--target "):
                idx = rest.find(" ", 9)
                if idx == -1:
                    target = rest[9:]; rest = ""
                else:
                    target = rest[9:idx]; rest = rest[idx + 1:].strip()
            send_task(target, rest)
        else:
            print("未知命令: {}".format(line))


def main():
    global SERVER, PORT, AUTH, DEVICE, WORKDIR, TASKS_DIR, OUTPUT_DIR
    args = build_argparser().parse_args()
    SERVER = args.server
    PORT = args.port
    AUTH = args.auth
    DEVICE = args.device
    if args.workdir:
        WORKDIR = args.workdir
    TASKS_DIR = os.path.join(WORKDIR, "tasks")
    OUTPUT_DIR = os.path.join(WORKDIR, "output")
    for _d in (WORKDIR, TASKS_DIR, OUTPUT_DIR):
        try:
            os.makedirs(_d, exist_ok=True)
        except OSError:
            pass

    cmd = args.cmd

    if cmd == "status":
        print(json.dumps(get_status(), ensure_ascii=False, indent=2))
        return
    if cmd == "list":
        st = get_status()
        print("在线: {}".format(
            [d for d, m in st.get("devices", {}).items()
             if m.get("online")]))
        print("队列: {}".format(st.get("queue_per_device", {})))
        return
    if cmd == "task":
        tgt = args.target or "all"
        prompt = " ".join(args.prompt).strip() if args.prompt else ""
        if not prompt:
            print("usage: client.py task --target <device|all> <prompt>")
            return
        send_task(tgt, prompt)
        return

    # 常驻模式
    print("=== DeepSeekSync v3 启动 ===")
    print("  服务器 : {}:{}".format(SERVER, PORT))
    print("  设备名 : {}".format(DEVICE))
    print("  workdir: {}".format(WORKDIR))
    print("  任务落盘: {}/".format(TASKS_DIR))
    print("  产物监视上传: {}/".format(OUTPUT_DIR))

    # 启动线程
    sync_file_list()  # 启动时先拉一次最近产物
    t_ws = threading.Thread(target=ws_main, daemon=True)
    t_watch = threading.Thread(target=watch_output_dir, daemon=True)
    t_ws.start()
    t_watch.start()

    if args.headless:
        try:
            while not STOP.is_set():
                STOP.wait(60)
        except KeyboardInterrupt:
            pass
        return
    try:
        interactive()
    finally:
        STOP.set()


if __name__ == "__main__":
    main()