下载 key 到同目录下的 keys.txt

#!/usr/bin/env python3
import argparse
import json
import os
import queue
import random
import ssl
import threading
import time
import urllib.error
import urllib.request


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Check OpenAI API keys and list available models per key."
    )
    parser.add_argument(
        "--input",
        default="keys.txt",
        help="Input txt file, one key per line. Default: keys.txt.",
    )
    parser.add_argument(
        "--output",
        default="valid_models.txt",
        help="Output txt file for valid keys and their models.",
    )
    parser.add_argument(
        "--bad-output",
        default="",
        help="Optional txt file for invalid keys and errors.",
    )
    parser.add_argument("--concurrency", type=int, default=64)
    parser.add_argument("--queue-size", type=int, default=4096)
    parser.add_argument("--timeout", type=float, default=12.0)
    parser.add_argument("--max-retries", type=int, default=3)
    parser.add_argument("--backoff-base", type=float, default=0.5)
    parser.add_argument("--backoff-max", type=float, default=8.0)
    parser.add_argument("--log-every", type=int, default=5000)
    parser.add_argument("--flush-every", type=int, default=1000)
    parser.add_argument(
        "--base-url",
        default=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com"),
    )
    parser.add_argument(
        "--organization",
        default=os.environ.get("OPENAI_ORG") or os.environ.get("OPENAI_ORGANIZATION"),
    )
    parser.add_argument("--project", default=os.environ.get("OPENAI_PROJECT"))
    parser.add_argument(
        "--extra-header",
        action="append",
        default=[],
        help='Extra header, format "Name: Value". Can be repeated.',
    )
    return parser.parse_args()


def normalize_key(line: str) -> str | None:
    stripped = line.strip()
    if not stripped or stripped.startswith("#"):
        return None
    return stripped.split()[0]


def parse_extra_headers(items: list[str]) -> dict[str, str]:
    headers: dict[str, str] = {}
    for item in items:
        if ":" not in item:
            raise ValueError(f"Invalid --extra-header value: {item}")
        name, value = item.split(":", 1)
        name = name.strip()
        value = value.strip()
        if not name:
            raise ValueError(f"Invalid --extra-header name: {item}")
        headers[name] = value
    return headers


def compute_sleep(attempt: int, backoff_base: float, backoff_max: float, retry_after: str | None) -> float:
    if retry_after:
        try:
            return min(backoff_max, float(retry_after))
        except ValueError:
            pass
    jitter = random.random() * 0.2
    return min(backoff_max, backoff_base * (2 ** attempt)) + jitter


def fetch_models_for_key(
        key: str,
        base_url: str,
        base_headers: dict[str, str],
        timeout: float,
        max_retries: int,
        backoff_base: float,
        backoff_max: float,
        ssl_context: ssl.SSLContext,
) -> tuple[list[str] | None, str, str]:
    headers = dict(base_headers)
    headers["Authorization"] = f"Bearer {key}"
    url = f"{base_url}/v1/models"

    for attempt in range(max_retries + 1):
        try:
            req = urllib.request.Request(url, headers=headers, method="GET")
            with urllib.request.urlopen(req, timeout=timeout, context=ssl_context) as resp:
                status = resp.getcode()
                body = resp.read()
            if status == 200:
                try:
                    payload = json.loads(body)
                except json.JSONDecodeError:
                    return None, "invalid_json", ""
                data = payload.get("data")
                if not isinstance(data, list):
                    return None, "invalid_response", ""
                models = sorted(
                    {item.get("id") for item in data if isinstance(item, dict) and item.get("id")}
                )
                return models, "ok", ""
            if status in (401, 403):
                return None, "unauthorized", f"http_{status}"
            if status == 429 or status >= 500:
                if attempt < max_retries:
                    time.sleep(compute_sleep(attempt, backoff_base, backoff_max, None))
                    continue
                return None, "retry_exhausted", f"http_{status}"
            return None, "http_error", f"http_{status}"
        except urllib.error.HTTPError as exc:
            status = exc.code
            retry_after = exc.headers.get("Retry-After") if exc.headers else None
            if status in (401, 403):
                return None, "unauthorized", f"http_{status}"
            if status == 429 or status >= 500:
                if attempt < max_retries:
                    time.sleep(compute_sleep(attempt, backoff_base, backoff_max, retry_after))
                    continue
                return None, "retry_exhausted", f"http_{status}"
            return None, "http_error", f"http_{status}"
        except urllib.error.URLError as exc:
            if attempt < max_retries:
                time.sleep(compute_sleep(attempt, backoff_base, backoff_max, None))
                continue
            return None, "network_error", str(exc)
        except Exception as exc:  # noqa: BLE001
            return None, "exception", str(exc)

    return None, "retry_exhausted", "unknown"


class Stats:
    def __init__(self, log_every: int) -> None:
        self.log_every = log_every
        self.lock = threading.Lock()
        self.processed = 0
        self.ok = 0
        self.bad = 0
        self.error = 0
        self.total = 0
        self.start = time.time()

    def set_total(self, total: int) -> None:
        with self.lock:
            self.total = total

    def update(self, status: str) -> None:
        with self.lock:
            self.processed += 1
            if status == "ok":
                self.ok += 1
            elif status == "unauthorized":
                self.bad += 1
            else:
                self.error += 1
            if self.log_every > 0 and self.processed % self.log_every == 0:
                elapsed = max(time.time() - self.start, 0.001)
                rate = self.processed / elapsed
                if self.total > 0:
                    remaining = max(self.total - self.processed, 0)
                    eta = remaining / rate if rate > 0 else 0.0
                    pct = (self.processed / self.total) * 100
                    print(
                        f"processed={self.processed}/{self.total} "
                        f"({pct:.1f}%) ok={self.ok} bad={self.bad} "
                        f"error={self.error} rate={rate:.1f}/s eta={eta:.0f}s",
                        flush=True,
                    )
                else:
                    print(
                        f"processed={self.processed} ok={self.ok} "
                        f"bad={self.bad} error={self.error} rate={rate:.1f}/s",
                        flush=True,
                    )


def worker(
        key_queue: queue.Queue,
        result_queue: queue.Queue,
        stats: Stats,
        base_url: str,
        base_headers: dict[str, str],
        timeout: float,
        max_retries: int,
        backoff_base: float,
        backoff_max: float,
        ssl_context: ssl.SSLContext,
) -> None:
    while True:
        key = key_queue.get()
        try:
            if key is None:
                return
            models, status, detail = fetch_models_for_key(
                key=key,
                base_url=base_url,
                base_headers=base_headers,
                timeout=timeout,
                max_retries=max_retries,
                backoff_base=backoff_base,
                backoff_max=backoff_max,
                ssl_context=ssl_context,
            )
            result_queue.put((key, models, status, detail))
            stats.update(status)
        finally:
            key_queue.task_done()


def writer(
        result_queue: queue.Queue,
        output_path: str,
        bad_output_path: str | None,
        flush_every: int,
) -> None:
    out_fp = open(output_path, "w", encoding="utf-8")
    bad_fp = open(bad_output_path, "w", encoding="utf-8") if bad_output_path else None
    written = 0
    bad_written = 0
    try:
        while True:
            item = result_queue.get()
            try:
                if item is None:
                    return
                key, models, status, detail = item
                if status == "ok" and models is not None:
                    out_fp.write(f"{key}\t{','.join(models)}\n")
                    written += 1
                    if flush_every > 0 and written % flush_every == 0:
                        out_fp.flush()
                elif bad_fp is not None:
                    bad_fp.write(f"{key}\t{status}\t{detail}\n")
                    bad_written += 1
                    if flush_every > 0 and bad_written % flush_every == 0:
                        bad_fp.flush()
            finally:
                result_queue.task_done()
    finally:
        out_fp.flush()
        out_fp.close()
        if bad_fp is not None:
            bad_fp.flush()
            bad_fp.close()


def main() -> int:
    args = parse_args()
    if args.concurrency <= 0:
        raise SystemExit("--concurrency must be > 0")
    if args.queue_size <= 0:
        raise SystemExit("--queue-size must be > 0")

    base_url = args.base_url.rstrip("/")
    base_headers = {"Accept": "application/json"}
    if args.organization:
        base_headers["OpenAI-Organization"] = args.organization
    if args.project:
        base_headers["OpenAI-Project"] = args.project
    if args.extra_header:
        base_headers.update(parse_extra_headers(args.extra_header))

    key_queue: queue.Queue = queue.Queue(maxsize=args.queue_size)
    result_queue: queue.Queue = queue.Queue(maxsize=args.queue_size)
    stats = Stats(log_every=args.log_every)
    ssl_context = ssl.create_default_context()

    bad_output = args.bad_output.strip() or None
    writer_thread = threading.Thread(
        target=writer,
        args=(result_queue, args.output, bad_output, args.flush_every),
        daemon=True,
    )
    writer_thread.start()

    print(
        "start "
        f"input={args.input} output={args.output} bad_output={bad_output or '-'} "
        f"concurrency={args.concurrency} queue_size={args.queue_size} "
        f"timeout={args.timeout} max_retries={args.max_retries}",
        flush=True,
    )

    workers: list[threading.Thread] = []
    for _ in range(args.concurrency):
        t = threading.Thread(
            target=worker,
            args=(
                key_queue,
                result_queue,
                stats,
                base_url,
                base_headers,
                args.timeout,
                args.max_retries,
                args.backoff_base,
                args.backoff_max,
                ssl_context,
            ),
            daemon=True,
        )
        t.start()
        workers.append(t)

    queued = 0
    with open(args.input, "r", encoding="utf-8", errors="ignore") as handle:
        for line in handle:
            key = normalize_key(line)
            if not key:
                continue
            key_queue.put(key)
            queued += 1
            if args.log_every > 0 and queued % args.log_every == 0:
                print(f"queued={queued}", flush=True)

    stats.set_total(queued)
    print(f"queued_total={queued}", flush=True)

    for _ in range(args.concurrency):
        key_queue.put(None)

    key_queue.join()
    for t in workers:
        t.join()

    result_queue.join()
    result_queue.put(None)
    writer_thread.join()

    elapsed = max(time.time() - stats.start, 0.001)
    print(
        f"done processed={stats.processed} ok={stats.ok} bad={stats.bad} "
        f"error={stats.error} seconds={elapsed:.1f}",
        flush=True,
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

关联 https://linux.do/t/topic/1496409/26

又到年终汇报时间，又要搞 word 格式，又要 ppt 形式述职，下面总结一下经验可以给佬们参考。

1. 先根据每个月的月总按照公司模板生成年总需要填写的内容，这里我用的 Google AI Studio - Gemini 3 pro。没有月总的可以看看自己 Git 提交记录找找今年做了什么。

2. 使用 Z.ai 生成 PPT。

这里导出的 PPTX 打开后，会发现样式有变化。

可以使用先导出 PDF，之后使用 Python 脚本将 PDF 转为图片，图片再转为 PPTX，这样就解决样式变化的问题了。

1. 先安装库

pip install pymupdf python-pptx

2. 创建文件 pdf2pptx.py，内容：

import fitz  # PyMuPDF
from pptx import Presentation
from pptx.util import Inches
import io
import os

def pdf_to_pptx(pdf_path, pptx_path, dpi=300):
    """
    将 PDF 转换为 PPTX（每一页作为一张全屏图片）
    :param pdf_path: 输入 PDF 文件路径
    :param pptx_path: 输出 PPTX 文件路径
    :param dpi: 渲染清晰度，300 已经非常清晰，如果文件太大可以降到 200
    """
    # 1. 打开 PDF
    pdf_doc = fitz.open(pdf_path)
    prs = Presentation()

    # 2. 获取 PDF 第一页的尺寸，并设置为 PPT 的幻灯片尺寸
    # PDF 默认单位是点 (points), 1 inch = 72 points
    first_page = pdf_doc[0]
    width_pts, height_pts = first_page.rect.width, first_page.rect.height
    
    prs.slide_width = Inches(width_pts / 72)
    prs.slide_height = Inches(height_pts / 72)

    print(f"开始转换: {pdf_path}")
    print(f"总页数: {len(pdf_doc)}")

    # 3. 逐页处理
    for page_num in range(len(pdf_doc)):
        page = pdf_doc.load_page(page_num)
        
        # 渲染页面为图片 (设置缩放倍数以提高清晰度)
        zoom = dpi / 72
        mat = fitz.Matrix(zoom, zoom)
        pix = page.get_pixmap(matrix=mat, alpha=False)
        
        # 将图片存入内存流中，避免产生临时文件
        img_stream = io.BytesIO(pix.tobytes("png"))
        
        # 添加一张空白幻灯片 (6 是空白布局)
        slide_layout = prs.slide_layouts[6]
        slide = prs.slides.add_slide(slide_layout)
        
        # 将图片插入幻灯片，铺满全屏
        slide.shapes.add_picture(img_stream, 0, 0, width=prs.slide_width, height=prs.slide_height)
        
        print(f"正在处理第 {page_num + 1} 页...")

    # 4. 保存文件
    prs.save(pptx_path)
    pdf_doc.close()
    print(f"转换完成！已保存至: {pptx_path}")

if __name__ == "__main__":
    # === 使用说明 ===
    # 将下面的文件名替换为你实际的文件名
    input_pdf = "input.pdf" 
    output_pptx = "output.pptx"

    if os.path.exists(input_pdf):
        pdf_to_pptx(input_pdf, output_pptx)
    else:
        print(f"错误：找不到文件 {input_pdf}")

3. 将 pdf 文件重名为 input.pdf，然后执行：

python pdf2pptx.py

4. 输出文件 output.pptx。可以再让 AI 帮你输出一份口述草稿配合起来就行了。

有尝试过在线 ILovePDF 网站去转，但是样式还是有问题。

试过 NotebookLM 生成 PPT，效果一般般，没想到 Z.ai 生成的效果还挺好看的

说是单 key 能用十次具体自己研究

注册机源码如下

 import requests
import random
import string
import re
import time
import urllib.parse

def generate_random_email():
    """生成随机8位前缀的邮箱"""
    prefix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
    return f"{prefix}@rccg-clf.org" def send_passwordless_init(email):
    """第一步：发送验证码"""
    url = "https://auth.privy.io/api/v1/passwordless/init"

    headers = {
        "accept": "application/json",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "cache-control": "no-cache",
        "content-type": "application/json",
        "origin": "https://beta.gatewayz.ai",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "privy-app-id": "cmg8fkib300g3l40dbs6autqe",
        "privy-ca-id": ,
        "privy-client": "react-auth:3.0.1",
        "privy-ui": "t",
        "referer": "https://beta.gatewayz.ai/",
        "sec-ch-ua": '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "sec-fetch-storage-access": "active",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0"
    }

    data = {"email": email}

    try:
        response = requests.post(url, headers=headers, json=data)
        return response.json()
    except Exception as e:
        return {"error": str(e)}

def check_email(email):
    """查询邮箱收到的邮件"""
    encoded_email = urllib.parse.quote(email, safe='')
    url = f"https://mail.chatgpt.org.uk/api/emails?email={encoded_email}"

    headers = {
        "accept": "*/*",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "cache-control": "no-cache",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "referer": f"https://mail.chatgpt.org.uk/{email}",
        "sec-ch-ua": '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0"
    }

    try:
        response = requests.get(url, headers=headers)
        return response.json()
    except Exception as e:
        return {"error": str(e)}

def extract_code(email_content):
    """从邮件内容中提取验证码""" match = re.search(r'\b(\d{6})\b', email_content)
    if match:
        return match.group(1)
    return None def wait_for_code(email, max_attempts=10, interval=2):
    """第二步：循环查询邮件直到获取验证码""" print(f"\n[2] 开始查询邮件 (最多{max_attempts}次，间隔{interval}秒)...")

    for attempt in range(1, max_attempts + 1):
        print(f"    第{attempt}次查询...", end=" ")

        mail_result = check_email(email)

        if mail_result.get("success"):
            emails = mail_result.get("data", {}).get("emails", [])
            if emails:
                for mail in emails:
                    if "privy" in mail.get("from_address", "").lower():
                        content = mail.get("content", "")
                        code = extract_code(content)
                        if code:
                            print(f"成功!")
                            return code
                print("未找到验证码邮件")
            else:
                print("暂无邮件")
        else:
            print(f"查询失败")

        if attempt < max_attempts:
            time.sleep(interval)

    return None def authenticate(email, code):
    """第三步：使用验证码登录获取token"""
    url = "https://auth.privy.io/api/v1/passwordless/authenticate"

    headers = {
        "accept": "application/json",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "cache-control": "no-cache",
        "content-type": "application/json",
        "origin": "https://beta.gatewayz.ai",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "privy-app-id": "cmg8fkib300g3l40dbs6autqe",
        "privy-ca-id": ,
        "privy-client": "react-auth:3.0.1",
        "referer": "https://beta.gatewayz.ai/",
        "sec-ch-ua": '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "cross-site",
        "sec-fetch-storage-access": "active",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0"
    }

    data = {
        "email": email,
        "code": code,
        "mode": "login-or-sign-up"
    }

    try:
        response = requests.post(url, headers=headers, json=data)
        return response.json()
    except Exception as e:
        return {"error": str(e)}

def create_api_key(auth_result, max_retries=3):
    """第四步：使用token创建API Key"""
    url = "https://beta.gatewayz.ai/api/auth"

    headers = {
        "accept": "*/*",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "cache-control": "no-cache",
        "content-type": "application/json",
        "origin": "https://beta.gatewayz.ai",
        "pragma": "no-cache",
        "priority": "u=1, i",
        "referer": "https://beta.gatewayz.ai/onboarding",
        "sec-ch-ua": '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0"
    }

    cookies = {
        "privy-token": auth_result.get("token", ""),
        "privy-session": "t"
    }

    data = {
        "user": auth_result.get("user", {}),
        "token": auth_result.get("token", ""),
        "auto_create_api_key": True,
        "is_new_user": auth_result.get("is_new_user", True),
        "has_referral_code": False,
        "referral_code": None,
        "privy_user_id": auth_result.get("user", {}).get("id", ""),
        "trial_credits": 10
    }

    for attempt in range(1, max_retries + 1):
        try:
            response = requests.post(url, headers=headers, cookies=cookies, json=data, timeout=30)
            result = response.json()
            if result.get("success") or "api_key" in result:
                return result
            # 如果返回错误但不是超时，直接返回 if "error" in result and "timeout" not in result.get("error", "").lower():
                return result
            # 如果是超时或其他错误，继续重试 if attempt < max_retries:
                print(f"    尝试 {attempt}/{max_retries} 失败，3秒后重试...")
                time.sleep(3)
        except requests.exceptions.Timeout:
            if attempt < max_retries:
                print(f"    请求超时 ({attempt}/{max_retries})，3秒后重试...")
                time.sleep(3)
            else:
                return {"error": "Request timeout after retries"}
        except Exception as e:
            if attempt < max_retries:
                print(f"    请求失败 ({attempt}/{max_retries}): {str(e)}，3秒后重试...")
                time.sleep(3)
            else:
                return {"error": str(e)}
    
    return {"error": "Max retries reached"}

def run():
    """运行完整流程""" print("=" * 50)
    print("Gatewayz 自动注册/登录")
    print("=" * 50)

    # 第一步：生成邮箱并发送验证码
    email = generate_random_email()
    print(f"\n[1] 生成邮箱: {email}")

    result = send_passwordless_init(email)
    if not result.get("success"):
        print(f"    发送验证码失败: {result}")
        return None print(f"    发送验证码成功!")

    # 第二步：获取验证码
    code = wait_for_code(email, max_attempts=10, interval=2)
    if not code:
        print(f"\n获取验证码失败，请手动查看: https://mail.chatgpt.org.uk/{email}")
        return None print(f"    验证码: {code}")

    # 第三步：登录认证 print(f"\n[3] 正在登录认证...")
    auth_result = authenticate(email, code)

    if "error" in auth_result:
        print(f"    登录失败: {auth_result}")
        return None if "token" not in auth_result:
        print(f"    登录失败: {auth_result}")
        return None print(f"    登录成功!")

    # 第四步：创建API Key print(f"\n[4] 正在创建API Key...")
    api_result = create_api_key(auth_result)

    if "error" in api_result:
        print(f"    创建API Key失败: {api_result}")
        return None if not api_result.get("success"):
        print(f"    创建API Key失败: {api_result}")
        return None

    api_key = api_result.get("api_key", "")
    if not api_key:
        print(f"    创建API Key失败: 未返回API Key")
        return None print(f"    创建成功!")

    # 保存API Key到文件 with open("api_keys.txt", "a", encoding="utf-8") as f:
        f.write(f"{api_key}\n")

    # 输出结果 print("\n" + "=" * 50)
    print("账号创建成功!")
    print("=" * 50)
    print(f"邮箱: {email}")
    print(f"用户ID: {api_result.get('user_id', 'N/A')}")
    print(f"Privy用户ID: {api_result.get('privy_user_id', 'N/A')}")
    print(f"试用积分: {api_result.get('credits', 'N/A')}")
    print(f"订阅状态: {api_result.get('subscription_status', 'N/A')}")
    print(f"试用到期时间: {api_result.get('trial_expires_at', 'N/A')}")
    print(f"\nAPI Key:\n{api_key}")
    print(f"\n已保存到: api_keys.txt")
    print("=" * 50)

    return {
        "email": email,
        "user_id": api_result.get('user_id'),
        "privy_user_id": api_result.get('privy_user_id'),
        "api_key": api_key,
        "credits": api_result.get('credits'),
        "subscription_status": api_result.get('subscription_status'),
        "trial_expires_at": api_result.get('trial_expires_at')
    }

if __name__ == "__main__":
    run()