smy-skills/quark-sign-skill/scripts/quark_sign.py

#!/usr/bin/env python3
"""
夸克网盘自动签到脚本

用法:
    python quark_sign.py --url "你的夸克URL" [--check]
    python quark_sign.py --config ~/.config/quark_sign.json
"""
import argparse
import json
import os
import sys
from urllib.parse import urlparse, parse_qs, unquote

# 尝试导入 requests，如果没有则使用 urllib
HAS_REQUESTS = True
try:
    import requests
except ImportError:
    HAS_REQUESTS = False
    import urllib.request
    import urllib.parse
    import urllib.error
    import ssl


class QuarkSign:
    BASE_URL = "https://drive-m.quark.cn/1/clouddrive/capacity/growth"

    def __init__(self, url: str):
        self.params = self._extract_params(url)
        if not self.params.get("kps") or not self.params.get("sign"):
            raise ValueError("URL 中缺少 kps 或 sign 参数，请检查 URL 是否完整")

    def _extract_params(self, url: str) -> dict:
        parsed = urlparse(url)
        params = parse_qs(parsed.query)
        return {k: unquote(v[0]) for k, v in params.items()}

    def _build_query(self, extra: dict = None) -> dict:
        q = {
            "pr": "ucpro",
            "fr": "android",
            "kps": self.params.get("kps"),
            "sign": self.params.get("sign"),
            "vcode": self.params.get("vcode", "1780291057288"),
        }
        if extra:
            q.update(extra)
        return q

    def _post(self, endpoint: str, data: dict = None, params: dict = None) -> dict:
        url = f"{self.BASE_URL}/{endpoint}"
        query = self._build_query(params)

        if HAS_REQUESTS:
            resp = requests.post(url, json=data or {}, params=query, timeout=10)
            return resp.json()
        else:
            # 使用 urllib  fallback
            full_url = url + "?" + urllib.parse.urlencode(query)
            payload = json.dumps(data or {}).encode("utf-8")
            req = urllib.request.Request(
                full_url,
                data=payload,
                headers={"Content-Type": "application/json"},
                method="POST",
            )
            ctx = ssl.create_default_context()
            with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
                return json.loads(resp.read().decode("utf-8"))

    def _get(self, endpoint: str, params: dict = None) -> dict:
        url = f"{self.BASE_URL}/{endpoint}"
        query = self._build_query(params)

        if HAS_REQUESTS:
            resp = requests.get(url, params=query, timeout=10)
            return resp.json()
        else:
            full_url = url + "?" + urllib.parse.urlencode(query)
            req = urllib.request.Request(full_url, method="GET")
            ctx = ssl.create_default_context()
            with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
                return json.loads(resp.read().decode("utf-8"))

    def sign(self) -> dict:
        """执行签到"""
        return self._post("sign", data={"sign_cyclic": True})

    def info(self) -> dict:
        """获取签到信息"""
        return self._get("sign", params={"_size": 1})

    def reward_info(self) -> dict:
        """获取奖励记录"""
        return self._get("reward_record", params={"_size": 10})


def format_reward(reward_bytes: int) -> str:
    if reward_bytes >= 1024 ** 3:
        return f"{reward_bytes / (1024 ** 3):.2f} GB"
    return f"{reward_bytes / (1024 ** 2):.2f} MB"


def main():
    parser = argparse.ArgumentParser(description="夸克网盘自动签到")
    parser.add_argument("--url", help="夸克网盘分享/签到 URL（包含 kps 和 sign 参数）")
    parser.add_argument("--config", "-c", help="配置文件路径（JSON 格式）")
    parser.add_argument("--check", action="store_true", help="仅检查签到状态，不执行签到")
    parser.add_argument("--info", action="store_true", help="显示签到信息")
    args = parser.parse_args()

    # 获取 URL
    url = args.url
    if not url and args.config:
        with open(os.path.expanduser(args.config)) as f:
            cfg = json.load(f)
            url = cfg.get("url")
    if not url:
        # 尝试环境变量
        url = os.environ.get("QUARK_SIGN_URL")
    if not url:
        # 尝试读取默认配置（skill 内置）
        script_dir = os.path.dirname(os.path.abspath(__file__))
        default_cfg = os.path.join(script_dir, "..", "config.json")
        if os.path.exists(default_cfg):
            with open(default_cfg) as f:
                cfg = json.load(f)
                url = cfg.get("url")
    if not url:
        print("❌ 请提供 URL：--url、--config 或环境变量 QUARK_SIGN_URL")
        print("   提示：可在 skill 目录下创建 config.json 写入默认 URL")
        sys.exit(1)

    try:
        qs = QuarkSign(url)
    except ValueError as e:
        print(f"❌ {e}")
        sys.exit(1)

    if args.info:
        print("📋 获取签到信息...")
        info = qs.info()
        print(json.dumps(info, indent=2, ensure_ascii=False))
        return

    if args.check:
        print("🔍 检查签到状态...")
        info = qs.info()
        data = info.get("data", {})
        if data.get("sign_daily"):
            print("✅ 今日已签到")
        else:
            print("⏳ 今日未签到")
        return

    # 执行签到
    print("🚀 执行签到...")
    result = qs.sign()

    status = result.get("status")
    code = result.get("code")
    msg = result.get("message", "")
    data = result.get("data", {})

    if data and status == 200:
        reward = data.get("sign_daily_reward", 0)
        print(f"✅ 签到成功！获得 {format_reward(reward)}")
    elif "repeat" in msg.lower() or code == 44210:
        print("ℹ️ 今日已签到过，无需重复签到")
    else:
        print(f"❌ 签到失败: {msg} (code={code}, status={status})")
        sys.exit(1)


if __name__ == "__main__":
    main()