sign.py

#sign-0.11
#!/usr/bin/env python3
# coding: utf-8
"""
post_with_tracks_and_logging.py

在原有功能基础上，增加每次请求/响应的完整记录：
- 追加 JSONL 每行一条：包含 request.headers/request.body, response.headers/response.body, token, track_id, timestamp, duration
- 同时保留简短的 results.log 输出
注意：会记录明文 session token，请妥善保护输出文件或启用掩码选项。
"""


import re
import sys
import time
import uuid
import random
import json
from typing import List, Set, Tuple, Optional
import requests
from datetime import datetime

INPUT_DEFAULT = "session_token_all.txt"
UNIQUE_TOKEN_OUT = "unique_tokens.txt"
UNIQUE_TRACK_OUT = "unique_track_ids.txt"
LOG_FILE = "results.log"
DETAILED_JSONL = "detailed_results.jsonl"
URL = "https://discount.wxpapp.wechatpay.cn/txbbs-mall/coupon/deliveryfreewithdrawalcoupon"

# 基础 headers（动态插入 session-token, X-Track-Id）
HEADERS_BASE = {
    "Host": "discount.wxpapp.wechatpay.cn",
    "Connection": "keep-alive",
    "X-Page": "pages/gift/index",
    "xweb_xhr": "1",
    "X-Module-Name": "mmpaytxbbsmp",
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/132.0.0.0 Safari/537.36 "
        "MicroMessenger/7.0.20.1781(0x6700143B) "
        "NetType/WIFI MiniProgramEnv/Windows "
        "WindowsWechat/WMPF WindowsWechat(0x63090a13) "
        "UnifiedPCWindowsWechat(0xf2541113) XWEB/16771"
    ),
    "Content-Type": "application/json",
    "X-Appid": "wxdb3c0e388702f785",
    "Accept": "*/*",
    "Sec-Fetch-Site": "cross-site",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Dest": "empty",
    "Referer": "https://servicewechat.com/wxdb3c0e388702f785/92/page-frame.html",
    "Accept-Encoding": "gzip, deflate, br",
    "Accept-Language": "zh-CN,zh;q=0.9",
}

MIN_TOKEN_LEN = 40

def gen_track_id(prefix: str = "T") -> str:
    return prefix + uuid.uuid4().hex.upper()

def load_file(path: str) -> str:
    try:
        with open(path, "r", encoding="utf-8", errors="ignore") as f:
            return f.read()
    except FileNotFoundError:
        print(f"[ERROR] 输入文件未找到: {path}")
        #0.07add 不退出程序，由调用方决定
        return None  
        #0.07del sys.exit(1)

def save_list_to_file(items: List[str], path: str):
    with open(path, "w", encoding="utf-8") as f:
        for it in items:
            f.write(it + "\n")

def append_log(line: str):
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(line + "\n")

def append_jsonl(obj: dict, path: str = DETAILED_JSONL):
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(obj, ensure_ascii=False) + "\n")

# 提取 token & track id（同之前实现）
def extract_tokens_from_text(text: str) -> List[str]:
    tokens_found = []
    seen = set()
    patterns = [
        re.compile(r'header:session-token\t([A-Za-z0-9_\-]+)', re.IGNORECASE),
        re.compile(r'session-token[:=]\s*([A-Za-z0-9_\-]+)', re.IGNORECASE),
        re.compile(r'"session_token"\s*:\s*"([^"]+)"', re.IGNORECASE),
        re.compile(r'response_json:.*?session_token[:=]\s*([A-Za-z0-9_\-]+)', re.IGNORECASE),
        re.compile(r'session_token[:=]\s*([A-Za-z0-9_\-]+)', re.IGNORECASE),
    ]
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        for p in patterns:
            m = p.search(line)
            if m:
                tok = m.group(1).strip()
                if len(tok) >= MIN_TOKEN_LEN and tok not in seen:
                    tokens_found.append(tok)
                    seen.add(tok)
    # 兜底
    fallback = re.compile(r'\b([A-Za-z0-9_\-]{' + str(MIN_TOKEN_LEN) + r',300})\b')
    for m in fallback.finditer(text):
        tok = m.group(1)
        if tok not in seen:
            tokens_found.append(tok)
            seen.add(tok)
    return tokens_found

def extract_track_ids_from_text(text: str) -> List[str]:
    tracks_found = []
    seen = set()
    patterns = [
        re.compile(r'X-Track-Id[:=\t ]+([A-Za-z0-9\-_]+)', re.IGNORECASE),
        re.compile(r'X-Track-Id"\s*:\s*"([^"]+)"', re.IGNORECASE),
    ]
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        for p in patterns:
            m = p.search(line)
            if m:
                tid = m.group(1).strip()
                if len(tid) >= 8 and tid not in seen:
                    tracks_found.append(tid)
                    seen.add(tid)
    fallback = re.compile(r'\bT([A-Fa-f0-9]{32})\b')
    for m in fallback.finditer(text):
        full = "T" + m.group(1).upper()
        if full not in seen:
            tracks_found.append(full)
            seen.add(full)
    return tracks_found

def mask_token(tok: str, keep_prefix: int = 6, keep_suffix: int = 4) -> str:
    """可选掩码：保留前 keep_prefix 和后 keep_suffix，其余替换为 *"""
    if tok is None:
        return None
    if len(tok) <= (keep_prefix + keep_suffix + 2):
        return tok[:keep_prefix] + "..." 
    return tok[:keep_prefix] + ("*" * (len(tok) - keep_prefix - keep_suffix)) + tok[-keep_suffix:]

def send_and_record(idx: int, token: str, track_id: str, mask_tokens: bool = False, timeout: float = 12.0):
    # 构造 headers + body
    headers = HEADERS_BASE.copy()
    headers["session-token"] = token
    headers["X-Track-Id"] = track_id
    request_body = {}  # 目前我们发送空 json body
    
    t0 = time.time()
    try:
        r = requests.post(URL, headers=headers, json=request_body, timeout=timeout,verify=False)
        duration = time.time() - t0
        status = r.status_code
        resp_text = r.text
        resp_headers = dict(r.headers)
    except requests.RequestException as e:
        duration = time.time() - t0
        status = -1
        resp_text = f"REQUEST_ERROR: {e}"
        resp_headers = {}
    
    # 简短 console/log 输出（保持原有简洁行为）
    masked = mask_token(token) if mask_tokens else (token[:12] + "..." if len(token) > 12 else token)
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    brief = (resp_text[:300] + "...") if len(resp_text) > 300 else resp_text
    if status == -1:
        line = f"{ts} [{idx}] {masked} -> Track {track_id} -> ERROR -> {brief}"
    else:
        line = f"{ts} [{idx}] {masked} -> Track {track_id} -> Status {status} -> {brief}"
    print(line)
    append_log(line)

    
    # 详细记录对象（完整）
    record = {
        "timestamp": ts,
        "index": idx,
        "token": (mask_token(token) if mask_tokens else token),
        "track_id": track_id,
        "request": {
            "method": "POST",
            "url": URL,
            "headers": headers,
            "body": request_body
        },
        "response": {
            "status_code": status,
            "headers": resp_headers,
            "body": resp_text
        },
        "meta": {
            "duration_seconds": duration
        }
    }
    append_jsonl(record)
    #0.10
#    return status
    #0.11
    return resp_text

def main(input_path: str = INPUT_DEFAULT, delay_sec: float = 0.5, mask_tokens: bool = True):
    text = load_file(input_path)
    #0.07 add
    if text is None:
        print(f"[WARN] 文件 {input_path} 不存在，跳过本轮")
        return   # 或 continue，如果在循环里

    tokens = extract_tokens_from_text(text)
    tracks = extract_track_ids_from_text(text)
    print(f"共提取到 {len(tokens)} 个去重后 token（按出现顺序）。")
    print(f"共提取到 {len(tracks)} 个去重后 X-Track-Id（按出现顺序）。")
    if not tokens:
        print("未找到 token，退出。")
        return
    save_list_to_file(tokens, UNIQUE_TOKEN_OUT)
    save_list_to_file(tracks, UNIQUE_TRACK_OUT)
    print(f"已把唯一 token 保存到: {UNIQUE_TOKEN_OUT}")
    print(f"已把唯一 track id 保存到: {UNIQUE_TRACK_OUT}")
    append_log(f"=== Run at {time.strftime('%Y-%m-%d %H:%M:%S')} - total {len(tokens)} tokens from {input_path} ===")
    if tracks:
        append_log(f"=== total {len(tracks)} track ids extracted ===")
    for idx, token in enumerate(tokens, start=1):
        if tracks:
            track_id = tracks[(idx - 1) % len(tracks)]
        else:
            track_id = gen_track_id()
        resp_text=send_and_record(idx, token, track_id, mask_tokens=mask_tokens)
        if delay_sec > 0:
            time.sleep(delay_sec)

    print("全部完成。简短日志:", LOG_FILE, "详细 JSONL:", DETAILED_JSONL)

    
    #0.10->
#0.11del    if status == 200:  # HTTP 200 表示签到成功

    #0.11 add
    if "face_value"  in resp_text or "已在其它微信领取" in resp_text:
        print("[INFO] 检测到 'face_value' 或'已在其它微信领取'，签到成功 ✅")
        return True
    if delay_sec > 0:
        time.sleep(delay_sec)
        return False
    #0.10<-

    

"""0.06del
if __name__ == "__main__":
    infile = INPUT_DEFAULT
    delay = 0.5
    mask = True
    if len(sys.argv) >= 2:
        infile = sys.argv[1]
    if len(sys.argv) >= 3:
        try:
            delay = float(sys.argv[2])
        except ValueError:
            pass
    # 可用第三个参数传 "no-mask" 关闭 token 掩码（默认开启）
    if len(sys.argv) >= 4 and sys.argv[3].lower() in ("no-mask", "nomask", "false"):
        mask = False
    main(infile, delay, mask)
"""

#0.06add
def run_with_args(infile=None, delay=None, mask=None):
    """完整模拟直接运行 sign.py 时的行为"""
    # 默认值
    infile = infile if infile is not None else INPUT_DEFAULT
    delay = delay if delay is not None else 0.5
    mask = mask if mask is not None else True

    # 如果调用时没有参数，使用 sys.argv 来解析命令行参数（兼容直接运行）
    if infile == INPUT_DEFAULT and len(sys.argv) >= 2:
        infile = sys.argv[1]
    if delay == 0.5 and len(sys.argv) >= 3:
        try:
            delay = float(sys.argv[2])
        except ValueError:
            pass
    if mask is True and len(sys.argv) >= 4 and sys.argv[3].lower() in ("no-mask", "nomask", "false"):
        mask = False

#0.09    main(infile, delay, mask)
#add 0.09
    result = main(infile, delay, mask)
    return result


# 保留原有命令行入口
if __name__ == "__main__":
    run_with_args()