# Source: https://github.com/Small-tailqwq/ql_script/blob/master/kuro/ql_kuro.py # Raw: https://raw.githubusercontent.com/Small-tailqwq/ql_script/master/kuro/ql_kuro.py # Repo: Small-tailqwq/ql_script # Path: kuro/ql_kuro.py # UploadedAt: 2026-05-23T15:50:58Z # SHA256: 46013225510794598730a4bad8387ebfbcee626a372ce9a1e605ff7acbf18bac # Category: web版/抓包 # Evidence: web/H5关键词 + cookie/token/header # -*- coding: utf-8 -*- """ cron: 30 8 * * * new Env('库街区自动签到'); 基于 mxyooR/Kuro-autosignin (GPL-3.0) 精简适配青龙单文件版 上游版本: a334796 (2026-04-21) 最后更新: 2026-05-23 09:00 (UTC+8) """ import datetime import json import logging import os import random import socket import time import uuid from typing import Dict, List, Optional import requests try: from sendNotify import send except Exception: try: from notify import send except Exception: send = None TIMEOUT = 30 GAME_WUWA_ID = "3" GAME_PGR_ID = "2" SERVER_WUWA = "76402e5b20be2c39f095a152090afddc" SERVER_PGR = "1000" MAX_RETRIES = 3 class API: USER_MINE = "https://api.kurobbs.com/user/mineV2" USER_SIGN_IN = "https://api.kurobbs.com/user/signIn" USER_ROLE_LIST = "https://api.kurobbs.com/user/role/findRoleList" FORUM_LIST = "https://api.kurobbs.com/forum/list" FORUM_POST_DETAIL = "https://api.kurobbs.com/forum/getPostDetail" FORUM_LIKE = "https://api.kurobbs.com/forum/like" TASK_PROCESS = "https://api.kurobbs.com/encourage/level/getTaskProcess" TASK_SHARE = "https://api.kurobbs.com/encourage/level/shareTask" GAME_SIGN_IN = "https://api.kurobbs.com/encourage/signIn/v2" GAME_SIGN_RECORD = "https://api.kurobbs.com/encourage/signIn/queryRecordV2" def get_ip_address() -> str: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except socket.error: return "127.0.0.1" finally: s.close() class KuroSigner: def __init__(self, token_data: Dict[str, str]): self.token = token_data['token'] self.note = token_data.get('note', '未命名账号') self.ip = get_ip_address() self.devcode = str(uuid.uuid4()) self.distinct_id = str(uuid.uuid4()) self.session = requests.Session() self.user_id = None self.wuwa_role_id = None self.pgr_role_id = None self.messages = [] self.success = True def _log(self, msg: str): logging.info(f"[{self.note}] {msg}") self.messages.append(msg) def _req(self, url: str, method: str = 'POST', data: Optional[dict] = None, req_type: str = 'bbs') -> dict: headers = { "Accept": "*/*", "Accept-Language": "zh-CN,zh-Hans;q=0.9", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive", "Host": "api.kurobbs.com", "source": "ios", } if req_type == 'bbs': headers.update({ "lang": "zh-Hans", "User-Agent": "KuroGameBox/48 CFNetwork/1492.0.1 Darwin/23.3.0", "channelId": "1", "channel": "appstore", "version": "2.2.0", "model": "iPhone15,2", "osVersion": "17.3", "Content-Type": "application/x-www-form-urlencoded; charset=utf-8", "Cookie": f"user_token={self.token}", "Ip": self.ip, "distinct_id": self.distinct_id, "devCode": self.devcode, "token": self.token, }) elif req_type == 'game': headers.update({ "Accept": "application/json, text/plain, */*", "Sec-Fetch-Site": "same-site", "source": "ios", "Sec-Fetch-Mode": "cors", "Origin": "https://web-static.kurobbs.com", "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) KuroGameBox/2.2.0", "devCode": f"{self.ip}, Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) KuroGameBox/2.2.0", "token": self.token, }) elif req_type == 'user_info': headers.update({ "osversion": "Android", "countrycode": "CN", "ip": self.ip, "model": "2211133C", "source": "android", "lang": "zh-Hans", "version": "1.0.9", "versioncode": "1090", "content-type": "application/x-www-form-urlencoded", "user-agent": "okhttp/3.10.0", "devcode": self.devcode, "distinct_id": self.distinct_id, "token": self.token, }) try: if method == 'POST': resp = self.session.post(url, headers=headers, data=data, timeout=TIMEOUT).json() else: resp = self.session.get(url, headers=headers, timeout=TIMEOUT).json() return resp except Exception as e: return {"code": -1, "msg": f"请求异常: {str(e)}"} def init_user_data(self) -> bool: res = self._req(API.USER_MINE, req_type='user_info') if res.get('code') != 200: self._log(f"获取用户信息失败或Token已过期: {res.get('msg')}") self.success = False return False self.user_id = res['data']['mine']['userId'] res_wuwa = self._req(API.USER_ROLE_LIST, data={"gameId": GAME_WUWA_ID}, req_type='user_info') if res_wuwa.get('code') == 200 and res_wuwa.get('data'): self.wuwa_role_id = res_wuwa['data'][0].get('roleId') res_pgr = self._req(API.USER_ROLE_LIST, data={"gameId": GAME_PGR_ID}, req_type='user_info') if res_pgr.get('code') == 200 and res_pgr.get('data'): self.pgr_role_id = res_pgr['data'][0].get('roleId') return True def _get_sign_reward(self, game_name: str, game_id: str, server_id: str, role_id: str) -> Optional[str]: try: data = { "gameId": game_id, "serverId": server_id, "roleId": role_id, "userId": self.user_id, } res = self._req(API.GAME_SIGN_RECORD, data=data, req_type='game') if res.get('code') == 200 and res.get('data'): if isinstance(res['data'], list) and len(res['data']) > 0: return res['data'][0].get('goodsName') return None except Exception: return None def game_sign(self): month = datetime.datetime.now().strftime("%m") def _do_sign(game_name, game_id, server_id, role_id): if not role_id: self._log(f"[{game_name}] 未找到角色ID，跳过签到") return data = { "gameId": game_id, "serverId": server_id, "roleId": role_id, "userId": self.user_id, "reqMonth": month, } res = self._req(API.GAME_SIGN_IN, data=data, req_type='game') code = res.get('code') if code == 200: reward = self._get_sign_reward(game_name, game_id, server_id, role_id) msg = f"[{game_name}] 签到成功" if reward: msg += f"，奖励: {reward}" self._log(msg) elif code == 1511: reward = self._get_sign_reward(game_name, game_id, server_id, role_id) msg = f"[{game_name}] 今日已签到" if reward: msg += f"，奖励: {reward}" self._log(msg) elif code == 1513: self._log(f"[{game_name}] 签到失败：用户信息异常，即将禁用该账号") self.success = False elif code == 220: self._log(f"[{game_name}] 签到失败：登录已过期，请重新登录") self.success = False else: self._log(f"[{game_name}] 签到失败: {res.get('msg', '未知错误')} (code={code})") raise Exception(f"{game_name} 签到失败: {res.get('msg', '未知错误')}") _do_sign("鸣潮", GAME_WUWA_ID, SERVER_WUWA, self.wuwa_role_id) time.sleep(1) _do_sign("战双", GAME_PGR_ID, SERVER_PGR, self.pgr_role_id) def forum_tasks(self): res = self._req(API.USER_SIGN_IN, data={"gameId": "2"}, req_type='bbs') if res.get('code') == 200: self._log("论坛签到成功") time.sleep(1) res_tasks = self._req(API.TASK_PROCESS, data={"gameId": "0"}, req_type='bbs') if res_tasks.get('code') != 200: self._log("获取每日任务列表失败跳过") return tasks = res_tasks.get('data', {}).get('dailyTask', []) tasks_map = {t.get('remark'): t.get('process') for t in tasks} posts = [] if tasks_map.get("浏览3篇帖子", 1) == 0 or tasks_map.get("点赞5次", 1) == 0: res_posts = self._req(API.FORUM_LIST, data={"forumId": "9", "gameId": "3", "pageIndex": "1", "pageSize": "20", "searchType": "3", "timeType": "0"}, req_type='bbs') if res_posts.get('code') == 200: posts = res_posts.get('data', {}).get('postList', []) if tasks_map.get("浏览3篇帖子", 1) == 0 and posts: view_count = 0 for p in posts[:3]: self._req(API.FORUM_POST_DETAIL, data={"isOnlyPublisher": "0", "postId": p['postId'], "showOrderTyper": "2"}, req_type='bbs') view_count += 1 time.sleep(random.uniform(1, 2)) self._log(f"完成浏览帖子任务 ({view_count}/3)") if tasks_map.get("点赞5次", 1) == 0 and posts: like_count = 0 for p in posts[:5]: self._req(API.FORUM_LIKE, data={"forumId": 11, "gameId": 3, "likeType": 1, "operateType": 1, "postCommentId": "", "postCommentReplyId": "", "postId": p['postId'], "postType": 1, "toUserId": p['userId']}, req_type='bbs') like_count += 1 time.sleep(random.uniform(1, 2)) self._log(f"完成点赞帖子任务 ({like_count}/5)") if tasks_map.get("分享1次帖子", 1) == 0: self._req(API.TASK_SHARE, data={"gameId": 3}, req_type='bbs') self._log("完成分享帖子任务 (1/1)") def run(self): for attempt in range(MAX_RETRIES): try: if attempt > 0: delay = random.uniform(5, 15) self._log(f"将在 {delay:.1f} 秒后进行第 {attempt + 1}/{MAX_RETRIES} 次重试...") time.sleep(delay) if self.init_user_data(): self.game_sign() self.forum_tasks() if not self.success: raise Exception("签到执行失败") break except Exception as e: self._log(f"执行异常: {str(e)}") self.success = False if attempt < MAX_RETRIES - 1: continue else: self._log(f"重试 {MAX_RETRIES} 次后仍然失败，放弃签到") return "\n".join(self.messages) def get_tokens() -> List[Dict[str, str]]: raw = os.environ.get("KURO_TOKEN", "").strip() if not raw: return [] raw = raw.replace('&', '\n').replace(',', '\n') token_list = [] for line in raw.split('\n'): line = line.strip() if not line: continue parts = line.split('#', 1) token_val = parts[0].strip() note = parts[1].strip() if len(parts) > 1 else f"账号{len(token_list) + 1}" if token_val: token_list.append({'token': token_val, 'note': note}) return token_list def main(): logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s') tokens = get_tokens() if not tokens: logging.error("未找到 Token，请在环境变量中设置 'KURO_TOKEN'。") return logging.info(f"✅ 检测到 {len(tokens)} 个账号，开始执行库街区签到...\n" + "-" * 30) notify_content = [] for token_data in tokens: signer = KuroSigner(token_data) result_msg = signer.run() status_icon = "✅" if signer.success else "❌" header = f"【{signer.note}】{status_icon}" notify_content.append(header) notify_content.append(result_msg) notify_content.append("-" * 20) final_content = "\n".join(notify_content) logging.info(f"\n{final_content}") if send: send("库街区签到", final_content) if __name__ == '__main__': main()