From 90a9d369b6fdf58c5458e864479a99f7b195dd33 Mon Sep 17 00:00:00 2001 From: sar Date: Mon, 5 Jan 2026 16:13:47 +0800 Subject: [PATCH] first commit --- autoDabing.py | 450 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 450 insertions(+) create mode 100644 autoDabing.py diff --git a/autoDabing.py b/autoDabing.py new file mode 100644 index 0000000..60bd323 --- /dev/null +++ b/autoDabing.py @@ -0,0 +1,450 @@ +import json +import random +import string +from typing import Optional, Dict, Tuple +import time +import httpx +import re +import requests +import logging + +from anyio import sleep + +logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + +# --- 配置信息 --- +BASE_URL = "https://mail.copy.qzz.io/api" +AUTH_TOKEN = 'ks4cPQfAvMQI1iqhFCDJTGpuc5ez6B95Iaf9SzA47MP0LiH5Pv7urHjjvVsHJlCY' + +# --- 公共变量 --- +headers = { + 'Authorization': 'Bearer ' + AUTH_TOKEN +} + + +def extract_verification_link(content): + """从邮件内容提取验证链接""" + match = re.search(r'href="(https://services\.sheerid\.com/verify/[^"]+emailToken=[^"]+)"', content) + if match: + return match.group(1).replace('&', '&') + + match = re.search(r'https://services\.sheerid\.com/verify/[^\s<>"]+emailToken=\d+', content) + if match: + return match.group(0) + + return None + +def generate_new_email(length: int = 8, domain_index: int = 0) -> str | None: + """ + 向 API 请求生成一个新的临时邮箱地址。 + + Args: + length (int): 邮箱地址的用户名长度。 + domain_index (int): 使用的域名索引。 + + Returns: + str | None: 成功时返回邮箱地址字符串，失败时返回 None。 + """ + # 1. 明确操作意图，并记录请求参数 + operation = "生成新邮箱地址" + endpoint = f"{BASE_URL}/generate" + params = {'length': length, 'domainIndex': domain_index} + + logger.info(f"开始操作: {operation} (参数: {params})") + + try: + # 使用 params 参数，让 requests 自动处理 URL 查询字符串 + logger.debug(f"发送 GET 请求至: {endpoint}") + response = requests.get(endpoint, headers=headers, params=params) + + # 2. 关联请求与响应 + logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}") + # 检查 HTTP 错误状态码 + response.raise_for_status() + + # 尝试解析 JSON + data = response.json() + logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}") + + email = data.get('email') + if email: + # 3. 增加成功日志 + logger.info(f"成功 {operation}，获取到邮箱: {email}") + return email + else: + logger.error(f"操作 '{operation}' 失败: 响应 JSON 中缺少 'email' 键。") + return None + except requests.exceptions.HTTPError as err: + # 4. 丰富错误上下文 + logger.error(f"操作 '{operation}' 失败，发生 HTTP 错误: {err}") + # 在 debug 级别记录完整的服务器响应，避免在 INFO/ERROR 级别刷屏 + logger.debug(f"服务器错误响应全文: {err.response.text}") + return None + + except requests.exceptions.RequestException as err: + logger.critical(f"操作 '{operation}' 失败，发生严重请求错误 (如网络问题): {err}") + return None + + except json.JSONDecodeError: + logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。") + logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'") + return None + + +def get_mail_detail_from_mailbox(email_address: str, email_id: int) -> str | None: + """ + 从指定的邮箱中获取所有邮件，并提取出 SheerID 验证链接。 + Args: + email_address (str): 要查询的临时邮箱地址。 + auth_token (str): 用于 API 认证的 Bearer Token。 + Returns: + list[str]: 一个包含所有找到的 SheerID 链接的列表。 + 如果找不到链接或发生错误，则返回一个空列表。 + """ + base_url = "https://mail.copy.qzz.io/api" + endpoint = f"{base_url}/email/{email_id}" + operation = f"获取邮箱 <{email_address}> 内的 SheerID 链接" + logger.info(f"开始操作: {operation}") + try: + # --- 发送 API 请求 --- + response = requests.get(endpoint, headers=headers) + + + + # --- 解析 JSON --- + data = response.json() + logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}") + extract_verification_link(data.get('html_content')) + return data.get('html_content') + + except requests.exceptions.HTTPError as err: + logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}") + logger.debug(f"服务器错误响应全文: {err.response.text}") + return None # 发生错误时返回空列表 + + except requests.exceptions.RequestException as err: + logger.critical(f"操作 '{operation}' 失败: 发生严重请求错误 (如网络问题): {err}") + return None + + except json.JSONDecodeError: + logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。") + logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'") + return None + + +def get_sheerid_links_from_mailbox(email_address: str) -> str | None: + """ + 从指定的邮箱中获取所有邮件，并提取出 SheerID 验证链接。 + Args: + email_address (str): 要查询的临时邮箱地址。 + auth_token (str): 用于 API 认证的 Bearer Token。 + Returns: + list[str]: 一个包含所有找到的 SheerID 链接的列表。 + 如果找不到链接或发生错误，则返回一个空列表。 + """ + base_url = "https://mail.copy.qzz.io/api" + endpoint = f"{base_url}/emails" + + params = { + "mailbox": email_address + } + + operation = f"获取邮箱 <{email_address}> 内的 SheerID 链接" + logger.info(f"开始操作: {operation}") + try: + # --- 发送 API 请求 --- + response = requests.get(endpoint, headers=headers, params=params) + response.raise_for_status() + logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}") + # --- 解析 JSON --- + data = response.json() + logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}") + if not data[0].get('sender') == "Verify@SheerID.com": + return None + id = data[0].get('id') + logger.debug(id) + content = get_mail_detail_from_mailbox(email_address, id) + url = extract_verification_link(content) + return url + + except requests.exceptions.HTTPError as err: + logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}") + logger.debug(f"服务器错误响应全文: {err.response.text}") + return None # 发生错误时返回空列表 + + except requests.exceptions.RequestException as err: + logger.critical(f"操作 '{operation}' 失败: 发生严重请求错误 (如网络问题): {err}") + return None + + except json.JSONDecodeError: + logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。") + logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'") + return None + + +def delete_mailbox(email_address: str) -> bool: + """ + 通过 API 删除一个临时邮箱地址（邮箱）。 + Args: + email_address (str): 要删除的邮箱地址。 + auth_token (str): 用于 API 认证的 Bearer Token。 + Returns: + bool: 如果成功删除则返回 True，否则返回 False。 + """ + # 稍微修改了函数名，因为 API 端点是 /mailboxes，更准确 + operation = f"删除邮箱 <{email_address}>" + logger.info(f"开始操作: {operation}") + # 1. 输入验证 + if not email_address: + logger.warning("尝试删除一个空的邮箱地址，操作已取消。") + return False + endpoint = f"{BASE_URL}/mailboxes" + params = { + 'address': email_address + } + + try: + # --- 发送 API 请求 --- + response = requests.delete(endpoint, headers=headers, params=params) + + logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}") + response.raise_for_status() # 如果状态码是 4xx 或 5xx，将在此处引发 HTTPError + # 2. 检查成功响应 + # 成功的 DELETE 请求通常返回 204 No Content (无响应体) + if response.status_code == 204: + logger.info(f"成功 {operation} (状态码 204)。") + return True + + # 如果返回 200 OK 并带有 JSON 体 + data = response.json() + logger.debug(f"成功 {operation}，收到的 JSON 详情: \n{json.dumps(data, indent=2)}") + logger.info(f"成功 {operation} (状态码 {response.status_code})。") + return True + except requests.exceptions.HTTPError as err: + logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}") + # 使用 debug 级别记录详细响应，避免刷屏 + logger.debug(f"服务器错误响应全文: {err.response.text}") + return False + + except requests.exceptions.RequestException as err: + logger.critical(f"操作 '{operation}' 失败: 发生网络等严重请求错误: {err}") + return False + + except json.JSONDecodeError: + logger.error(f"操作 '{operation}' 失败: 服务器返回了成功的状态码，但响应不是有效的 JSON。") + logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'") + return False + + +class SheerIDVerifier: + def __init__(self, verification_id: str,program_id:str) -> None: + self.program_id = program_id + self.verification_id = verification_id + self.device_fingerprint = self._generate_device_fingerprint() + self.http_client = httpx.Client(timeout=30.0) + + def __del__(self): + if hasattr(self, "http_client"): + self.http_client.close() + + @staticmethod + def _generate_device_fingerprint() -> str: + chars = '0123456789abcdef' + return ''.join(random.choice(chars) for _ in range(32)) + + @staticmethod + def normalize_url(url: str) -> str: + """规范化 URL（保留原样）""" + return url + + @staticmethod + def parse_verification_id(url: str) -> Optional[str]: + match = re.search(r"verificationId=([a-f0-9]+)", url, re.IGNORECASE) + if match: + return match.group(1) + return None + + @staticmethod + def parse_program_id(url: str) -> Optional[str]: + match = re.search(r"verify/([a-f0-9]+)/", url, re.IGNORECASE) + if match: + return match.group(1) + return None + + def _sheerid_request( + self, method: str, url: str, body: Optional[Dict] = None + ) -> Tuple[Dict, int]: + """发送 SheerID API 请求""" + headers = { + 'host': 'services.sheerid.com', + 'sec-ch-ua-platform': '"Windows"', + 'sec-ch-ua': f'"Chromium";v="142", "Google Chrome";v="142", "Not_A Brand";v="99"', + 'clientversion': '2.157.0', + 'sec-ch-ua-mobile': '?0', + 'clientname': 'jslib', + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36', + 'accept': 'application/json', + 'content-type': 'application/json', + 'origin': 'https://services.sheerid.com', + 'sec-fetch-site': 'same-origin', + 'sec-fetch-mode': 'cors', + 'sec-fetch-dest': 'empty', + 'referer': f'https://services.sheerid.com/verify/{self.program_id}/?verificationId={self.verification_id}', + 'accept-encoding': 'gzip, deflate, br, zstd', + 'accept-language': 'en-US,en-GB;q=0.9,en;q=0.8', + 'priority': 'u=1, i', + } + + try: + response = self.http_client.request( + method=method, url=url, json=body, headers=headers + ) + try: + data = response.json() + except Exception: + data = response.text + return data, response.status_code + except Exception as e: + logger.error(f"SheerID 请求失败: {e}") + raise + + def verify(self) -> Dict: + try: + + email_address = generate_new_email(10,0) + time.sleep(30) + submit_data, submit_status = self._sheerid_request( + "POST", + f"https://services.sheerid.com/rest/v2/verification/{self.verification_id}/step/collectMilitaryStatus", + {'status': 'VETERAN'}, + ) + + if submit_status != 200: + raise Exception(f"提交状态 失败 (状态码 {submit_status}): {submit_data}") + if submit_data.get("currentStep") != "collectInactiveMilitaryPersonalInfo": + error_msg = ", ".join(submit_data.get("errorIds", ["Unknown error"])) + raise Exception(f"提交状态 错误: {error_msg}") + + + submissionUrl = submit_data.get("submissionUrl") + verificationId = re.search(r"verificationId=([a-f0-9]+)", submissionUrl, re.IGNORECASE) + + random_string = ''.join(random.choices(string.ascii_uppercase, k=5)) + + verify_body = { + "firstName": "KELLY " + random_string, + "lastName": "BURKHART", + "birthDate": "1981-05-14", + "email": email_address, + "phoneNumber": "", + "organization": { + "id": 4070, + "name": "Army" + }, + "dischargeDate": "2025-10-29", + "locale": "en-US", + "country": "US", + "metadata": { + "marketConsentValue": False, + "refererUrl": f"https://services.sheerid.com/verify/{self.program_id}/?verificationId={self.verification_id}", + "verificationId": verificationId, + "flags": "{\"doc-upload-considerations\":\"default\",\"doc-upload-may24\":\"default\",\"doc-upload-redesign-use-legacy-message-keys\":false,\"docUpload-assertion-checklist\":\"default\",\"include-cvec-field-france-student\":\"not-labeled-optional\",\"org-search-overlay\":\"default\",\"org-selected-display\":\"default\"}", + "submissionOptIn": "By submitting the personal information above, I acknowledge that my personal information is being collected under the privacy policy of the business from which I am seeking a discount, and I understand that my personal information will be shared with SheerID as a processor/third-party service provider in order for SheerID to confirm my eligibility for a special offer. Contact OpenAI Support for further assistance at support@openai.com" + } + } + + info_data, info_status = self._sheerid_request( + "POST", + submissionUrl, + verify_body, + ) + + if info_status != 200: + raise Exception(f"提交信息 失败 (状态码 {info_status}): {info_data}") + if info_data.get("currentStep") == "error": + error_msg = ", ".join(info_data.get("errorIds", ["Unknown error"])) + raise Exception(f"提交信息 错误: {error_msg}") + + time.sleep(120) + + finish_verifying_url = get_sheerid_links_from_mailbox(email_address) + while True: + if finish_verifying_url: + break + time.sleep(120) + finish_verifying_url = get_sheerid_links_from_mailbox(email_address) + logger.debug("完成验证 " + finish_verifying_url) + + final_data, final_status = self._sheerid_request( + "POST", + finish_verifying_url, + ) + if final_status != 200: + raise Exception(f"完成验证 失败 (状态码 {submit_status}): {submit_data}") + + logger.debug(f"完成验证 {json.dumps(final_data, indent=2)}") + + time.sleep(30) + delete_mailbox(email_address) + # 不做状态轮询，直接返回等待审核 + return { + "success": True, + "pending": True, + "verification_id": self.verification_id, + "redirect_url": final_data.get("redirectUrl"), + "status": final_data, + } + + + except Exception as e: + logger.error(f"❌ 验证失败: {e}") + return {"success": False, "message": str(e), "verification_id": self.verification_id} + + +def main(): + """主函数 - 命令行界面""" + import sys + + print("=" * 60) + print("SheerID 学生身份验证工具 (Python版)") + print("=" * 60) + print() + + if len(sys.argv) > 1: + url = sys.argv[1] + else: + url = input("请输入 SheerID 验证 URL: ").strip() + + if not url: + print("❌ 错误: 未提供 URL") + sys.exit(1) + + verification_id = SheerIDVerifier.parse_verification_id(url) + program_id = SheerIDVerifier.parse_program_id(url) + if not verification_id and not program_id: + print("❌ 错误: 无效的验证 ID 格式") + sys.exit(1) + + print(f"✅ 解析到验证 ID: {verification_id}\n✅ 解析到验证 ID: {program_id}", end="\n") + + verifier = SheerIDVerifier(verification_id,program_id) + result = verifier.verify() + + print() + print("=" * 60) + print("验证结果:") + print("=" * 60) + print(f"状态: {'✅ 成功' if result['success'] else '❌ 失败'}") + print(f"消息: {result['message']}") + if result.get("redirect_url"): + print(f"跳转 URL: {result['redirect_url']}") + print("=" * 60) + + return 0 if result["success"] else 1 + +if __name__ == '__main__': + exit(main()) \ No newline at end of file