🤔 前言

结合别帐号密码登陆的，用AI修改了下，支持帐号密码登陆了

📦 部署方法

宝塔面板添加定时任务

计划任务 → 添加任务：

任务类型：Python脚本
任务名称：女仆论坛签到
执行周期：每天 00:05
执行用户：root 或 www
python环境：任何版本

📜 完整代码

python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
女仆论坛自动签到 - 账号密码登录版 (支持企业微信/钉钉推送)
适用于 bt.sb (Rhex 论坛系统)
"""
import json
import urllib.request
import urllib.error
import logging
import os
import hmac
import hashlib
import base64
import time
from datetime import datetime

# ═══════════════ 配置区 ═══════════════
BASE_URL = "https://bt.sb"
# 账号密码配置（建议通过环境变量设置）
USERNAME = os.environ.get("BBS_USERNAME", "你的账号或邮箱")
PASSWORD = os.environ.get("BBS_PASSWORD", "你的密码")
LOG_FILE = os.environ.get("BBS_LOG", "")

# Webhook 配置
# 企业微信 Webhook 地址
WECHAT_WEBHOOK = os.environ.get("WECHAT_WEBHOOK", "你的企业微信Webhook地址")
# 钉钉 Webhook 地址
DINGTALK_WEBHOOK = os.environ.get("DINGTALK_WEBHOOK", "你的钉钉Webhook地址")
# 钉钉加签密钥 (如果钉钉机器人开启了加签安全设置，请填写此处，否则留空)
DINGTALK_SECRET = os.environ.get("DINGTALK_SECRET", "")

def setup_logging():
    handlers = [logging.StreamHandler()]
    if LOG_FILE:
        handlers.append(logging.FileHandler(LOG_FILE, encoding="utf-8"))
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [签到] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        handlers=handlers,
    )

def login():
    """使用账号密码登录，获取认证信息（Cookie或Token）"""
    url = f"{BASE_URL}/api/auth/login"
    login_data = {
        "login": USERNAME,
        "password": PASSWORD,
        "captchaToken": "",
        "builtinCaptchaCode": "",
        "powNonce": "",
        "addonFields": {}
    }
    data = json.dumps(login_data).encode("utf-8")
    req = urllib.request.Request(url, data=data, headers={
        "Content-Type": "application/json",
        "User-Agent": "Mozilla/5.0 BBS-AutoCheckIn/1.0",
        "Origin": BASE_URL,
        "Referer": f"{BASE_URL}/login",
    }, method="POST")
    
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            # 尝试解析响应
            response_body = resp.read().decode("utf-8")
            response_data = json.loads(response_body) if response_body else {}
            
            # 检查登录是否成功
            if resp.status != 200 or response_data.get("success") is False or response_data.get("ok") is False:
                error_msg = response_data.get("message") or response_data.get("error") or response_data.get("detail") or "登录失败"
                raise Exception(f"登录API返回错误: {error_msg}")
            
            # 从响应头获取Cookie
            cookie_header = resp.headers.get("Set-Cookie")
            if not cookie_header:
                # 如果响应头没有Cookie，尝试从响应体获取token
                token = response_data.get("token") or response_data.get("accessToken") or response_data.get("access_token")
                if token:
                    return f"Authorization: Bearer {token}"
                else:
                    raise Exception("登录响应中未找到认证信息(Cookie或Token)")
            
            # 提取有效的Cookie字段
            cookies = []
            for cookie in cookie_header.split(","):
                cookie = cookie.strip()
                if ";" in cookie:
                    cookies.append(cookie.split(";")[0])
                else:
                    cookies.append(cookie)
            
            return "; ".join(cookies)
            
    except urllib.error.HTTPError as e:
        raise Exception(f"登录HTTP错误 {e.code}: {e.reason}")
    except Exception as e:
        raise Exception(f"登录过程异常: {str(e)}")

def checkin(auth_header):
    """使用登录获得的认证信息执行签到"""
    url = f"{BASE_URL}/api/check-in"
    data = json.dumps({"action": "check-in"}).encode("utf-8")
    
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "Mozilla/5.0 BBS-AutoCheckIn/1.0",
        "Referer": BASE_URL,
        "Origin": BASE_URL,
    }
    
    # 根据认证信息类型设置请求头
    if auth_header.startswith("Authorization:"):
        headers["Authorization"] = auth_header.replace("Authorization: ", "")
    else:
        headers["Cookie"] = auth_header
    
    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
    
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            body = json.loads(resp.read().decode("utf-8"))
            payload = body.get("data", body)
            
            # 获取当前日期（用于消息显示）
            current_date = datetime.now().strftime("%Y-%m-%d")
            
            if payload.get("alreadyCheckedIn"):
                msg = f"📌 日期：{payload.get('date', current_date)} 已签到！"
            else:
                points = payload.get("points", "")
                streak = payload.get("currentStreak", 0)
                max_streak = payload.get("maxStreak", 0)
                msg = f"✅ 日期：{payload.get('date', current_date)} 签到成功！累计 {points} 女仆币 | 连续 {streak} 天 | 最长 {max_streak} 天！"
            
            logging.info(msg)
            push_notify(msg)
            return True, msg
            
    except urllib.error.HTTPError as e:
        current_date = datetime.now().strftime("%Y-%m-%d")
        if e.code in (401, 403):
            msg = f"❌ 日期：{current_date} 认证已失效（{e.code}），请检查账号密码！"
        else:
            msg = f"⚠️ 日期：{current_date} 签到异常 {e.code}: {e.reason}！"
        logging.error(msg)
        push_notify(msg)
        return False, msg
    except Exception as e:
        current_date = datetime.now().strftime("%Y-%m-%d")
        msg = f"❌ 日期：{current_date} 网络错误：{str(e)}！"
        logging.error(msg)
        push_notify(msg)
        return False, msg

def generate_dingtalk_sign(secret):
    """生成钉钉加签签名"""
    timestamp = str(round(time.time() * 1000))
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign

def send_wechat_webhook(msg):
    """发送消息到企业微信"""
    if not WECHAT_WEBHOOK or "key=" not in WECHAT_WEBHOOK:
        return
    try:
        payload = {
            "msgtype": "text",
            "text": {
                "content": msg
            }
        }
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(WECHAT_WEBHOOK, data=data, headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=10) as resp:
            res_body = json.loads(resp.read().decode("utf-8"))
            if res_body.get("errcode") != 0:
                logging.warning(f"📨 企业微信推送失败: {res_body.get('errmsg')}")
            else:
                logging.info("📨 企业微信推送成功")
    except Exception as e:
        logging.warning(f"📨 企业微信推送异常: {e}")

def send_dingtalk_webhook(msg):
    """发送消息到钉钉"""
    if not DINGTALK_WEBHOOK or "access_token=" not in DINGTALK_WEBHOOK:
        return
    try:
        webhook_url = DINGTALK_WEBHOOK
        if DINGTALK_SECRET:
            timestamp, sign = generate_dingtalk_sign(DINGTALK_SECRET)
            webhook_url = f"{DINGTALK_WEBHOOK}&timestamp={timestamp}&sign={sign}"

        payload = {
            "msgtype": "text",
            "text": {
                "content": msg
            },
            "at": {
                "isAtAll": False
            }
        }
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(webhook_url, data=data, headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=10) as resp:
            res_body = json.loads(resp.read().decode("utf-8"))
            if res_body.get("errcode") != 0:
                logging.warning(f"📨 钉钉推送失败: {res_body.get('errmsg')}")
            else:
                logging.info("📨 钉钉推送成功")
    except Exception as e:
        logging.warning(f"📨 钉钉推送异常: {e}")

def push_notify(msg):
    """统一推送入口"""
    send_wechat_webhook(msg)
    send_dingtalk_webhook(msg)

if __name__ == "__main__":
    setup_logging()
    logging.info("🕐 开始执行签到流程")
    
    # 检查账号密码配置
    if USERNAME == "在这里填写你的账号或邮箱" or not USERNAME:
        logging.error("❌ 请先配置 BBS_USERNAME 环境变量")
        exit(1)
    if PASSWORD == "在这里填写你的密码" or not PASSWORD:
        logging.error("❌ 请先配置 BBS_PASSWORD 环境变量")
        exit(1)
    
    try:
        # 1. 登录获取认证信息
        logging.info("🔐 正在登录...")
        auth_info = login()
        logging.info("✅ 登录成功，获取到认证信息")
        
        # 2. 使用认证信息执行签到
        logging.info("📝 正在执行签到...")
        success, result_msg = checkin(auth_info)
        
        # 3. 输出最终结果
        if success:
            logging.info("🎉 签到流程完成")
        else:
            logging.error("💥 签到流程失败")
            
    except Exception as e:
        error_msg = f"❌ 签到流程异常: {str(e)}"
        logging.error(error_msg)
        push_notify(error_msg)
        exit(1)