Showing Posts From
Python
OpenAI Videos API:Sora 2 视频生成 Python 调用指南
Python 调用示例 from openai import OpenAIclient = OpenAI(api_key="YOUR_API_KEY")# 提交视频生成任务(异步) job = client.videos.create( model="sora-2", prompt=""" A cinematic drone shot of Tokyo at night. Neon lights reflecting on wet streets. Ultra realistic. """, seconds=8, size="1280x720" )print(job.id)视频生成是异步任务,需要轮询状态: import timewhile True: result = client.videos.retrieve(job.id) print(result.status) if result.status == "completed": print(result.output_url) break elif result.status == "failed": print("生成失败") break time.sleep(5)图片转视频 已有参考图片时,使用 input_reference 保持人物和风格一致: job = client.videos.create( model="sora-2", prompt="The girl smiles and walks forward.", input_reference=open("girl.png", "rb"), seconds=8 )REST API curl https://api.openai.com/v1/videos \ -H "Authorization: Bearer YOUR_API_KEY" \ -F "model=sora-2" \ -F "prompt=A cute panda drinking coffee" \ -F "seconds=8"参数说明参数 说明model sora-2 或 sora-2-proprompt 视频内容描述(英文效果更好)input_reference 参考图片(可选,用于图生视频)seconds 视频时长:4、8、12 秒size 分辨率,如 1280x720、720x1280(竖屏)、1792x1024价格模型 价格Sora 2 约 $0.10 / 秒Sora 2 Pro 约 $0.30~$0.70 / 秒(随分辨率变化)生成 8 秒 1280x720 视频:Sora 2 约 $0.80,Sora 2 Pro 约 $2.40~5.60。 Prompt 写法参考 视频 prompt 建议包含:镜头类型、光线、风格、质量要求。 # 电影风格 A cinematic shot of a samurai walking through a bamboo forest. Golden sunlight. Slow motion. 35mm film. Ultra realistic.# 产品广告 A luxury mechanical watch rotating on a black reflective table. Studio lighting. 8K. Commercial quality.# 动漫风格 Anime style. Cherry blossoms falling. Girl running on a school campus. Beautiful sunset. Highly detailed.英文 prompt 效果稳定,中文可能出现理解偏差。 后端架构(完整平台) 如果需要支持多用户、异步进度追踪,典型架构: 用户请求 → FastAPI → Celery 任务队列 → OpenAI Videos API ↓ Redis 存储 job.id ↓ WebSocket 推送进度 → 前端 ↓ 完成后存储到 OSS/S3
EIP-1559 Gas 费计算与 Python 转账清仓脚本
EIP-1559 的 Gas 费有三个参数,容易写错导致交易失败或余额卡住。 三个参数的含义参数 含义baseFee 当前区块的基础 Gas 价格,由网络自动调整,不能低于这个值maxPriorityFeePerGas 给矿工的小费(Tip),影响打包速度maxFeePerGas 用户愿意支付的最高单价,节点用 max(baseFee + tip, maxFeePerGas) 结算实际每笔交易扣费:gasUsed × (baseFee + tip),多余部分退还。 常见错误写法 # 问题:maxFee 等于 baseFee + priority,baseFee 稍微上涨就失效 max_fee_per_gas = base_fee + priority_fee如果在你签名到广播之间的时间里,下一个区块的 baseFee 上涨,节点会拒绝交易(因为 maxFeePerGas < baseFee)。 推荐写法 base_fee = web3.eth.get_block("pending")["baseFeePerGas"]try: priority_fee = web3.eth.max_priority_fee except Exception: priority_fee = web3.to_wei(0.1, "gwei")# baseFee 乘以 2 留有余量(MetaMask 等钱包的标准做法) max_fee_per_gas = base_fee * 2 + priority_fee清仓转账(全额转出) 把钱包余额全部转出时,amount = balance - fee 看起来简单,但有几个坑: from web3 import Web3def transfer_all(web3, private_key, dst_address): account = web3.eth.account.from_key(private_key) from_address = account.address balance = web3.eth.get_balance(from_address) nonce = web3.eth.get_transaction_count(from_address, "pending") base_fee = web3.eth.get_block("pending")["baseFeePerGas"] try: priority_fee = web3.eth.max_priority_fee except Exception: priority_fee = web3.to_wei(0.1, "gwei") max_fee_per_gas = base_fee * 2 + priority_fee # estimate_gas 比写死 30000 更准确 gas_limit = web3.eth.estimate_gas({ "from": from_address, "to": dst_address, "value": 1 }) gas_limit = int(gas_limit * 1.2) fee = gas_limit * max_fee_per_gas # 额外留安全余量,防止 baseFee 微小变化导致不足 reserve = web3.to_wei(0.00000001, "ether") amount = balance - fee - reserve if amount <= 0: print("余额不足支付 Gas 费") return None tx = { "from": from_address, "to": Web3.to_checksum_address(dst_address), "value": amount, "gas": gas_limit, "maxFeePerGas": max_fee_per_gas, "maxPriorityFeePerGas": priority_fee, "nonce": nonce, "chainId": web3.eth.chain_id, "type": 2, } signed = web3.eth.account.sign_transaction(tx, private_key) tx_hash = web3.eth.send_raw_transaction(signed.raw_transaction) print(f"发送地址: {from_address}") print(f"转出: {web3.from_wei(amount, 'ether')} ETH") print(f"Gas 费: {web3.from_wei(fee, 'ether')} ETH") print(f"TX: {tx_hash.hex()}") return tx_hash.hex()关键细节 nonce 用 "pending" 而非 "latest":如果有未确认交易,用 "latest" 得到的 nonce 偏小,新交易会被拒绝。 gas_limit 用 estimateGas:普通 ETH 转账是 21000,但某些链可能不同,estimate_gas * 1.2 更稳妥。 chainId 用 web3.eth.chain_id 获取:不要硬编码,避免接错链。 确认当前链 ID print(web3.eth.chain_id)Gravity L1 Mainnet 的 Chain ID 是 127001,支持 EIP-1559。 两步计算更可靠 对于高精度要求的清仓场景,建议:先构造 value=0 估算 gasLimit 计算 fee = gasLimit × maxFeePerGas,加安全余量 签名广播 如果仍然报 insufficient funds(baseFee 变化导致),重新读取参数再签一次
EIP-1559 Gas 费计算:Python Web3 转账扣余额的正确姿势
做链上转账时,想把账户余额全部转走,需要提前扣除 Gas 费。EIP-1559 下的费用计算和普通 Legacy 交易有些区别,容易算错。 常见的错误写法 base_fee = w3.eth.get_block("latest")["baseFeePerGas"] priority_fee = w3.to_wei(0.1, "gwei") max_fee_per_gas = base_fee + priority_feegas_limit = 30000 fee = gas_limit * max_fee_per_gasamount = balance - fee这里有两个问题: 问题一:fee 计算偏高但不算错 gas_limit * max_fee_per_gas 是最坏情况下的费用上限,实际扣款是: actual_fee = gas_used × (base_fee + priority_fee)gas_used ≤ gas_limit,所以实际花费通常比预留值少。转账类交易 gas_used 基本等于 gas_limit,误差不大;合约调用误差可能较大。 问题二:maxFeePerGas 应该 > base_fee + priority_fee EIP-1559 规范里,maxFeePerGas 是你愿意支付的上限,真正矿工收到的是: effective_gas_price = base_fee + priority_fee如果下一个区块 base_fee 涨了,用 maxFeePerGas = base_fee + priority_fee 可能导致交易因 maxFeePerGas < 新 base_fee 而被拒绝。 正确写法 def send_all_balance(w3, private_key, to_address, chain_id): account = w3.eth.account.from_key(private_key) from_address = account.address balance = w3.eth.get_balance(from_address) nonce = w3.eth.get_transaction_count(from_address) # base_fee 加 20% 缓冲,防止下一块涨价 base_fee = w3.eth.get_block("latest")["baseFeePerGas"] base_fee_with_buffer = int(base_fee * 1.2) priority_fee = w3.to_wei(0.1, "gwei") max_fee_per_gas = base_fee_with_buffer + priority_fee gas_limit = 21000 # 纯转账固定 21000;合约调用需要 estimateGas # 预留的 Gas 费上限 max_gas_cost = gas_limit * max_fee_per_gas amount = balance - max_gas_cost if amount <= 0: print(f"余额不足:balance={w3.from_wei(balance, 'ether')} ETH") return None tx = { "from": from_address, "to": w3.to_checksum_address(to_address), "value": amount, "gas": gas_limit, "maxFeePerGas": max_fee_per_gas, "maxPriorityFeePerGas": priority_fee, "nonce": nonce, "chainId": chain_id, "type": 2, } signed = w3.eth.account.sign_transaction(tx, private_key) tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction) print(f"发送: {w3.from_wei(amount, 'ether')} ETH") print(f"Gas 预留: {w3.from_wei(max_gas_cost, 'ether')} ETH") print(f"TX: {tx_hash.hex()}") return tx_hash.hex()EIP-1559 费用结构字段 含义baseFeePerGas 协议自动设置,每笔交易必须支付,销毁maxPriorityFeePerGas 给矿工的小费,激励打包maxFeePerGas 用户愿意支付的上限实际每 Gas 费用 min(maxFeePerGas, baseFee + priorityFee)实际从账户扣除: actual_cost = gas_used × effective_gas_price其中 effective_gas_price = baseFee + priorityFee(不超过 maxFeePerGas)。 多余的 (maxFeePerGas - effectiveGasPrice) × gasUsed 会退还给发送方,但退还发生在交易上链后,所以发送前账户余额必须 ≥ gasLimit × maxFeePerGas。 纯转账 vs 合约调用 纯 ETH 转账 gas 固定是 21000,可以直接写死。合约调用要用 estimate_gas: gas_limit = w3.eth.estimate_gas({ "from": from_address, "to": contract_address, "data": encoded_data, }) gas_limit = int(gas_limit * 1.1) # 加 10% 缓冲合约执行失败时 gas 仍会被扣,所以建议先 call 验证不会 revert 再发送。 常见的链 gas_limit 不同链的标准转账 gas 不完全一样,普通 EVM 链一般是 21000,部分 L2 有所不同: GAS_LIMITS = { "ethereum": 21000, "polygon": 21000, "bsc": 21000, "gravity": 30000, # 某些链会高一些 }不确定的情况下用 estimate_gas 最安全。 一句话总结 EIP-1559 转账预留费用用 gasLimit × maxFeePerGas,maxFeePerGas 要比当前 baseFee 高出一定缓冲(一般 20%),防止下一块 baseFee 上涨导致交易失败。纯转账 gas 固定 21000,合约调用用 estimateGas。
EIP-1559 交易脚本的六个 Gas 坑:baseFee 突变到 nonce 冲突
写一个"清仓转账"的脚本——把地址里所有余额扫到目标地址,用 EIP-1559。核心代码大致这样: base_fee = w3.eth.get_block("latest")["baseFeePerGas"] priority_fee = w3.to_wei(0.1, "gwei") max_fee_per_gas = base_fee + priority_feegas_limit = 30000 fee = gas_limit * max_fee_per_gas amount = balance - feetx = { "to": w3.to_checksum_address(dst), "value": amount, "gas": gas_limit, "maxFeePerGas": max_fee_per_gas, "maxPriorityFeePerGas": priority_fee, "nonce": w3.eth.get_transaction_count(from_address), "chainId": CHAIN_ID, "type": 2, }跑起来经常报: insufficient funds for gas * price + value问题不是一处,是好几个细节叠加。 坑 1:maxFeePerGas 写死等于 baseFee + priority 节点校验时: balance >= value + gasLimit × maxFeePerGasbaseFee 每个区块都会浮动(EIP-1559 每次可变 ±12.5%)。你算完刚好够,几秒后新块 baseFee 涨了: max_fee < new_base_fee + priority要么交易过不了、要么"钱不够"。MetaMask 的做法是: max_fee_per_gas = base_fee * 2 + priority_feebase_fee * 2 是常规冗余系数——baseFee 一个区块最多涨 12.5%,2x 完全够缓冲。多花的部分节点会退还给你,只按 effectiveGasPrice 收。 坑 2:gasLimit 硬编码 普通转账 21000 够,Gravity 这类链要 30000。合约调用差别就更大了。别写死,先估算再加冗余: gas_limit = w3.eth.estimate_gas({ "from": from_address, "to": dst, "value": 1, }) gas_limit = int(gas_limit * 1.2)* 1.2 是通用的 20% 余量,避免 estimation 边缘 case。 坑 3:priority fee 硬编码 0.1 gwei 网络拥堵时 0.1 gwei 的小费根本进不了下一个块。用节点建议值: priority_fee = w3.eth.max_priority_fee或者查 feeHistory 取近期百分位: history = w3.eth.fee_history(5, "latest", [50]) priority_fee = max(history["reward"][-1][0], w3.to_wei(1, "gwei"))坑 4:nonce 用了 latest 默认: w3.eth.get_transaction_count(from_address) # 等价于 "latest"如果地址有未确认的 pending 交易,latest 会返回过时的 nonce,新交易和旧的撞车,两个都卡住。改成: nonce = w3.eth.get_transaction_count(from_address, "pending")坑 5:清仓转账没留冗余 amount = balance - fee 是理论极限。实际因为 baseFee 突变或 estimation 偏差,几乎必然凑不够。留一点小额缓冲: reserve = w3.to_wei(0.0001, "ether") amount = balance - fee - reserve坑 6:chainId 抄错 "chainId": 127001,链 ID 不匹配的交易节点直接拒。跟节点确认一次: print(w3.eth.chain_id)修正后的样板 base_fee = w3.eth.get_block("latest")["baseFeePerGas"] priority_fee = w3.eth.max_priority_fee max_fee_per_gas = base_fee * 2 + priority_feegas_limit = w3.eth.estimate_gas({ "from": from_address, "to": dst, "value": 1, }) gas_limit = int(gas_limit * 1.2)fee_ceiling = gas_limit * max_fee_per_gas reserve = w3.to_wei(0.0001, "ether") amount = balance - fee_ceiling - reserve if amount <= 0: raise RuntimeError("余额不足以支付 Gas + 冗余")tx = { "to": w3.to_checksum_address(dst), "value": amount, "gas": gas_limit, "maxFeePerGas": max_fee_per_gas, "maxPriorityFeePerGas": priority_fee, "nonce": w3.eth.get_transaction_count(from_address, "pending"), "chainId": w3.eth.chain_id, "type": 2, }一句话总结 EIP-1559 参数不是"算准就行",是"抗突变才稳"。maxFee = baseFee * 2 + priority、gas = estimate * 1.2、nonce = pending,三件事一起做,出问题的概率会低一个数量级。
Web3.py EIP-1559 交易 Gas 费计算:baseFee 波动与余额预留
EIP-1559 Gas 费结构 EIP-1559 交易有两个关键字段:maxFeePerGas:你愿意支付的最高单价(base fee + priority fee) maxPriorityFeePerGas:给矿工的小费("tip")节点在验证交易时检查: balance >= value + gasLimit × maxFeePerGas实际扣费是: gasUsed × (baseFeePerGas + priorityFeePerGas)其中 baseFeePerGas 由网络自动调整,gasUsed <= gasLimit。 典型实现(web3.py) from web3 import Web3w3 = Web3(Web3.HTTPProvider("https://mainnet-rpc.example.xyz"))def send_max(private_key, dst_address, chain_id): account = w3.eth.account.from_key(private_key) from_address = account.address balance = w3.eth.get_balance(from_address) nonce = w3.eth.get_transaction_count(from_address) # 动态读取 baseFee base_fee = w3.eth.get_block("latest")["baseFeePerGas"] # 用节点推荐的 priority fee,更可靠 priority_fee = w3.eth.max_priority_fee # maxFee 加 buffer,防 baseFee 上涨 max_fee_per_gas = base_fee * 2 + priority_fee # 估算 gas,比写死更安全 gas_limit = w3.eth.estimate_gas({ "from": from_address, "to": dst_address, "value": 1, }) gas_limit = int(gas_limit * 1.2) # 20% buffer fee = gas_limit * max_fee_per_gas # 预留少量 wei 防止因 baseFee 微小波动导致余额不足 reserve = w3.to_wei(0.000001, "ether") amount = balance - fee - reserve if amount <= 0: print(f"余额不足,balance={balance}, fee={fee}") return None tx = { "from": from_address, "to": Web3.to_checksum_address(dst_address), "value": amount, "gas": gas_limit, "maxFeePerGas": max_fee_per_gas, "maxPriorityFeePerGas": priority_fee, "nonce": nonce, "chainId": chain_id, "type": 2, } signed = w3.eth.account.sign_transaction(tx, private_key) tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction) return tx_hash.hex()常见坑 坑一:maxFee 太紧,baseFee 上涨就失败 # 错误写法:maxFee 刚好等于 baseFee + priority max_fee = base_fee + priority_fee # 一旦下一个区块 baseFee 上涨,交易立即失效改为: max_fee = base_fee * 2 + priority_fee # 留足余量,只多花极少的钱坑二:写死 gasLimit gas_limit = 21000 # ETH 原生转账是这个值,但其他链可能不同更安全的方式: gas_limit = w3.eth.estimate_gas({"from": from_addr, "to": dst, "value": 1}) gas_limit = int(gas_limit * 1.2)坑三:不预留 reserve 导致余额精确到 wei 后失败 节点验证 balance >= value + gasLimit × maxFeePerGas。如果 amount = balance - fee 算完后 baseFee 微涨,条件就不满足。 预留 0.000001 ETH(约 1,000,000,000,000 wei)几乎没有损失但能避免大量边缘失败。 坑四:priority_fee 写死 priority_fee = w3.to_wei(0.1, "gwei") # 可能太低导致交易迟迟不确认改为从节点获取推荐值: priority_fee = w3.eth.max_priority_fee # 节点返回当前合理的 tip
读取 CPU 型号信息:Windows 注册表与 Linux /proc/cpuinfo 方法汇总
CPU 型号信息(如 12th Gen Intel(R) Core(TM) i5-12400)的唯一真实来源是 CPUID 指令,操作系统在启动时读取 CPUID 并将结果缓存到系统结构中,之后所有查询都从这份缓存中读取。 Windows:注册表路径 CPU 信息存放在: HKEY_LOCAL_MACHINE\HARDWARE\DESCRIPTION\System\CentralProcessor\0关键字段:键名 示例值ProcessorNameString 12th Gen Intel(R) Core(TM) i5-12400VendorIdentifier GenuineIntel~MHz 2500命令行查询: reg query "HKEY_LOCAL_MACHINE\HARDWARE\DESCRIPTION\System\CentralProcessor\0"wmic 查询(更简洁): wmic cpu get name导出整个 CentralProcessor 分支: reg export "HKLM\HARDWARE\DESCRIPTION\System\CentralProcessor" cpu.regregedit 图形界面: 导航到上述路径 → 右键 → 导出。 Python 读取注册表 import winregkey_path = r"HARDWARE\DESCRIPTION\System\CentralProcessor\0" key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path)cpu_name, _ = winreg.QueryValueEx(key, "ProcessorNameString") vendor, _ = winreg.QueryValueEx(key, "VendorIdentifier") mhz, _ = winreg.QueryValueEx(key, "~MHz")print(cpu_name) # 12th Gen Intel(R) Core(TM) i5-12400Linux:/proc/cpuinfo cat /proc/cpuinfo | grep "model name" | head -1 # model name : 12th Gen Intel(R) Core(TM) i5-12400lscpu | grep "Model name" # Model name: 12th Gen Intel(R) Core(TM) i5-12400/proc/cpuinfo 是内核动态生成的虚拟文件,每次读取都从内核的 cpu_info 结构体生成,本质是 CPUID 的解析结果。 Python 读取: with open('/proc/cpuinfo') as f: for line in f: if line.startswith('model name'): print(line.split(':')[1].strip()) break注意:注册表信息可以被伪装 注册表里的 CPU 信息不是"唯一可信来源":虚拟机(VMware/VirtualBox/Hyper-V)可以在配置中修改 CPUID 返回值 驱动层可以拦截并修改 WMI 查询结果因此注册表/WMI 读取的 CPU 型号可能与物理 CPU 不一致,这是虚拟化平台的常见行为。 各方法对比方法 平台 特点wmic cpu get name Windows 最简单reg query Windows 可脚本化winreg (Python) Windows 程序读取lscpu Linux 格式化输出/proc/cpuinfo Linux 原始详情
BIP39 助记词生成 Solana 地址:SLIP-10 + Ed25519 全流程
BIP39 助记词生成 ETH 地址靠 secp256k1 + keccak256,走 Solana 就完全换了一套——Ed25519 + SLIP-0010,最后不做 hash,直接 Base58 编码公钥。 全流程 助记词 (Mnemonic) │ PBKDF2-HMAC-SHA512 ▼ Seed (64 字节) │ SLIP-0010 (Ed25519) ▼ Master Key │ 按路径 m/44'/501'/0'/0' 派生 ▼ Ed25519 私钥 (32 字节) │ Ed25519 公钥算法 ▼ 公钥 (32 字节) │ Base58 编码 ▼ Solana 地址和 BTC/ETH 唯一相同的只有第一步"助记词转 seed",之后全变了。 第一步:助记词 → seed 标准 BIP39,PBKDF2-HMAC-SHA512: password = 助记词字符串 salt = "mnemonic" + passphrase iters = 2048 output = 64 字节Python: from mnemonic import Mnemonicmnemo = Mnemonic("english") seed = mnemo.to_seed( "abandon abandon abandon abandon abandon abandon " "abandon abandon abandon abandon abandon about", passphrase="", ) print(seed.hex())第二步:seed → Master Key(SLIP-10 Ed25519) Solana 用 SLIP-0010(Ed25519 版本),和 BIP32 的 secp256k1 分道扬镳: I = HMAC-SHA512(key="ed25519 seed", data=seed) IL = master private key (32 字节) IR = master chain code (32 字节)第三步:派生路径 Solana 默认路径: m/44'/501'/0'/0'44':BIP44 501':Solana 的 coin type 0':account 0':change每一级都是 hardened derivation(' 表示 + 2³¹)。Ed25519 只支持 hardened 派生,非 hardened 会直接报错。 第四步:公钥 & 地址 私钥经 Ed25519 算出 32 字节公钥。地址 = Base58(公钥),就这么简单:没有 SHA-256 double hash 没有 RIPEMD-160 没有 keccak 没有 checksum(Base58 本身没自带校验,也没有像 BTC 那样附 4 字节 hash)Python 一把梭 bip_utils 把上面所有细节封好了: from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coinsmnemonic = ("abandon abandon abandon abandon abandon abandon " "abandon abandon abandon abandon abandon about")seed = Bip39SeedGenerator(mnemonic).Generate()bip44 = Bip44.FromSeed(seed, Bip44Coins.SOLANA) account = bip44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(0)print("地址:", account.PublicKey().ToAddress()) print("私钥:", account.PrivateKey().Raw().ToHex())Phantom / Solflare 里的第一个账号导出出来应该跟上面一样。 常见误区 Solana 地址和 ETH 地址长得像 其实完全不同:ETH 地址:40 字符 hex + 0x 前缀,checksum 在 EIP-55 里靠大小写实现 Solana 地址:Base58 编码的 32 字节公钥,长度 32~44 字符,无固定前缀Solana 私钥文件是 64 字节而不是 32 Solana 的钱包 JSON(Phantom 导出)通常是 64 字节:前 32 是私钥 seed、后 32 是公钥。生成时: priv32 = account.PrivateKey().Raw().ToBytes() pub32 = account.PublicKey().RawUncompressed().ToBytes()[-32:] keypair_bytes = priv32 + pub32 # 64 字节导入 solana-cli 或 Phantom 时用这个格式。 Ed25519 派生路径可以省略部分层级 Phantom 早期用 m/44'/501'/0'(只三层)而非 m/44'/501'/0'/0'——同一助记词在这两种路径下算出来的地址不同。发现"钱包地址对不上"时先检查是不是路径差异,试试:m/44'/501'/0' — Phantom "legacy" m/44'/501'/0'/0' — 现在的默认一句话总结 Solana 地址生成 = BIP39 seed → SLIP-10 Ed25519 派生 → Base58(公钥)。和 EVM 完全两套加密路径,别拿 ETH 的经验类比。用 bip_utils 三行搞定,别自己实现 Ed25519。
Python 跨平台获取 CPU 型号:py-cpuinfo / wmic / /proc/cpuinfo 全方案
问题背景 需要在 Python 代码里获取完整的 CPU 型号字符串,例如: Intel(R) Xeon(R) CPU E5-2680 v4 @ 2.40GHz不同平台读取方式不同,以下是各方案的完整写法。 推荐:py-cpuinfo(跨平台) pip install py-cpuinfofrom cpuinfo import get_cpu_infoinfo = get_cpu_info() print(info["brand_raw"]) # Intel(R) Xeon(R) CPU E5-2680 v4 @ 2.40GHzbrand_raw 是最原始、最完整的 CPU 商品名称。该库会自动根据平台选择 Windows 注册表、Linux /proc/cpuinfo、macOS sysctl 等数据源,兼容性最好,适合写跨平台工具。 Windows(无第三方库) 方案一:wmic import subprocesscpu = subprocess.check_output( "wmic cpu get name", shell=True ).decode("gbk").split("\n")[1].strip()print(cpu)注意编码用 gbk,Windows 中文环境下 cmd 输出默认是 GBK。 方案二:PowerShell CimInstance import subprocesscpu = subprocess.check_output( ["powershell", "(Get-CimInstance Win32_Processor).Name"] ).decode("utf-8").strip()print(cpu)PowerShell 输出是 UTF-8,更可靠,推荐用这个。 Linux 直接读 /proc/cpuinfo: with open("/proc/cpuinfo") as f: for line in f: if line.startswith("model name"): print(line.split(":", 1)[1].strip()) break多核机器 /proc/cpuinfo 会重复每个核的信息,读第一条就够了。 macOS import subprocesscpu = subprocess.check_output( ["sysctl", "-n", "machdep.cpu.brand_string"] ).decode().strip()print(cpu)不推荐:platform.processor() import platformprint(platform.processor()) # 很多系统输出:Intel64 Family 6 Model 79 Stepping 1, GenuineIntel # 或者直接为空字符串platform.processor() 不返回 CPU 商品名称,在 Linux 上通常是架构描述或空串,不适合用来显示给用户看的型号。 跨平台封装 from cpuinfo import get_cpu_infodef get_cpu_name() -> str: return get_cpu_info()["brand_raw"]用 py-cpuinfo 一行搞定,不需要为三个平台分别写判断逻辑。
PyTorch 读取 CSV 数据:pandas 转 Tensor、自定义 Dataset 与 DataLoader
快速读取(小数据) 先用 pandas 读 CSV,再一次性转成 Tensor: import pandas as pd import torchdf = pd.read_csv("house.csv")# 前 N-1 列是特征,最后一列是标签 x = torch.tensor(df.iloc[:, :-1].values, dtype=torch.float32) y = torch.tensor(df.iloc[:, -1].values, dtype=torch.float32)print(x.shape, y.shape) # torch.Size([N, features]) torch.Size([N])iloc[:, :-1] 取除最后列之外的所有列,iloc[:, -1] 取最后一列。 自定义 Dataset(正式训练推荐) 当数据量大或需要在线增强时,继承 Dataset 类实现懒加载: import pandas as pd import torch from torch.utils.data import Datasetclass CSVDataset(Dataset): def __init__(self, path): self.data = pd.read_csv(path) def __len__(self): return len(self.data) def __getitem__(self, index): row = self.data.iloc[index] x = torch.tensor(row[:-1].values, dtype=torch.float32) y = torch.tensor(row[-1], dtype=torch.float32) return x, y必须实现三个方法:__init__:加载/读取数据 __len__:返回样本总数 __getitem__:返回第 index 个样本DataLoader 分批读取 from torch.utils.data import DataLoaderdataset = CSVDataset("house.csv")loader = DataLoader( dataset, batch_size=32, shuffle=True, num_workers=4 # 多进程预取,Windows 下注意加 if __name__ == "__main__" )for x_batch, y_batch in loader: print(x_batch.shape) # [32, features] print(y_batch.shape) # [32]DataLoader 自动处理:打乱数据(shuffle=True) 按 batch_size 分批 多线程预取(num_workers)完整训练示例 import torch import torch.nn as nn from torch.utils.data import DataLoadermodel = nn.Sequential( nn.Linear(2, 16), nn.ReLU(), nn.Linear(16, 1) )optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) criterion = nn.MSELoss()dataset = CSVDataset("house.csv") loader = DataLoader(dataset, batch_size=8, shuffle=True)for epoch in range(100): for x, y in loader: pred = model(x).squeeze() loss = criterion(pred, y) optimizer.zero_grad() loss.backward() optimizer.step() if epoch % 10 == 0: print(f"Epoch {epoch}, Loss: {loss.item():.4f}")训练集与验证集分割 from torch.utils.data import random_splitdataset = CSVDataset("house.csv")train_size = int(0.8 * len(dataset)) val_size = len(dataset) - train_sizetrain_set, val_set = random_split(dataset, [train_size, val_size])train_loader = DataLoader(train_set, batch_size=32, shuffle=True) val_loader = DataLoader(val_set, batch_size=32)处理标签类型 回归任务用 float32,分类任务用 long: # 回归:y 是连续值 y = torch.tensor(row[-1], dtype=torch.float32)# 分类:y 是类别编号 y = torch.tensor(int(row[-1]), dtype=torch.long)分类时对应的 loss 是 nn.CrossEntropyLoss(),回归时用 nn.MSELoss()。
用 Python 构建高考志愿填报系统:位次换算与录取概率预测
为什么不用深度学习 这个场景是表格数据回归预测,不是图像识别。数据量有限(全国高校 × 省份 × 年份),LightGBM 等梯度提升模型在表格数据上通常比神经网络效果更好,且训练速度快。 技术栈 Python ├── Pandas 数据清洗和处理 ├── LightGBM 预测最低录取位次 ├── FastAPI 后端 API ├── SQLAlchemy 数据库 ORM └── Vue3 前端展示第一层:位次换算 真正决定录取的是位次,而非分数。同一个位次在不同年份对应的分数不同: import pandas as pd# 读取当年一分一段表 score_table = pd.read_csv("score_rank_2025.csv") # 结构:score, rank(分数→位次)def score_to_rank(score: int, year: int) -> int: df = score_table[score_table["year"] == year] row = df[df["score"] == score] if row.empty: # 取最接近的分数 row = df.iloc[(df["score"] - score).abs().argsort()[:1]] return int(row["rank"].values[0])位次越小录取概率越高(1 代表全省第一名)。 第二层:训练预测模型 历史录取数据结构:year school major province min_rank plan_count2023 清华 计算机 江苏 235 152024 清华 计算机 江苏 198 12import lightgbm as lgb from sklearn.model_selection import train_test_splitdf = pd.read_csv("admission_history.csv")features = ["year", "school_id", "major_id", "province_id", "plan_count"] target = "min_rank"X_train, X_test, y_train, y_test = train_test_split( df[features], df[target], test_size=0.2, random_state=42 )model = lgb.LGBMRegressor(n_estimators=500, learning_rate=0.05) model.fit(X_train, y_train)也可以替换为 XGBRegressor 或 CatBoostRegressor,效果相近。 第三层:计算录取概率 基于历史波动区间: def admission_probability(school_id, major_id, province_id, student_rank: int, year: int) -> float: # 预测今年最低录取位次 predicted_min_rank = model.predict([[year, school_id, major_id, province_id, plan_count]])[0] # 历史波动标准差 history = df[(df.school_id == school_id) & (df.major_id == major_id)] std = history["min_rank"].std() # 学生位次越小(越靠前)于预测线,概率越高 z = (predicted_min_rank - student_rank) / (std + 1) from scipy.stats import norm return float(norm.cdf(z))输出 0.0~1.0 的概率值。 冲/稳/保 分类 def classify(prob: float) -> str: if prob >= 0.8: return "保" # 稳上 elif prob >= 0.5: return "稳" # 有竞争力 else: return "冲" # 有风险但值得一试FastAPI 接口 from fastapi import FastAPIapp = FastAPI()@app.get("/predict") def predict(school: str, major: str, province: str, score: int, year: int): rank = score_to_rank(score, year) prob = admission_probability(school, major, province, rank, year) label = classify(prob) return { "rank": rank, "probability": round(prob, 3), "label": label }模型效果优化特征工程:加入招生人数变化趋势、学校综合排名、专业热度 位次分段:分段训练(Top 1000 / 1000~10000 / 10000+),各段波动规律不同 时序权重:近年数据给更高权重(sample_weight 参数) 多省合并训练:数据量小时跨省训练,加省份编码作为特征数据来源 历年录取数据可从各省教育考试院官网下载,或使用第三方爬虫整理的开源数据集(GitHub 搜索"高考录取数据")。确保数据包含:学校、专业、省份、年份、最低分、最低位次、计划人数。
矩阵运算的几何意义:变换、旋转与神经网络中的线性层
矩阵是"变换机器" 标量乘法 3 x 5 = 15 只是把一个数放大 3 倍。要对一个二维点进行缩放、旋转、镜像,一个数字不够用,需要矩阵。 矩阵乘以向量 = 对该点执行一次变换: import numpy as npA = np.array([[2, 0], [0, 2]]) # 缩放矩阵x = np.array([2, 3]) # 原始点 (2, 3)print(A @ x) # [4, 6] -> 放大 2 倍常见变换矩阵 缩放(Scaling): [sx 0 ] 把 x 缩放 sx 倍 [0 sy] 把 y 缩放 sy 倍旋转 angle(Rotation): [cos a -sin a] [sin a cos a]逆时针旋转 90 度: import math theta = math.pi / 2R = np.array([[math.cos(theta), -math.sin(theta)], [math.sin(theta), math.cos(theta)]])x = np.array([1, 0]) print(R @ x) # [0, 1] — (1,0) 逆时针旋转到 (0,1)镜像(y 轴翻转): [-1 0] [ 0 1]为什么矩阵乘法是那个规则 [1 2] [5] = [1x5 + 2x6] = [17] [3 4] x [6] [3x5 + 4x6] [39]不是人为规定,而是"两次变换合并"的自然结果:矩阵的每一行定义输出的一个维度,该维度由输入向量的线性组合得出。 矩阵乘法 = 变换的复合 两个矩阵连乘等于先做 B 变换再做 A 变换: A = np.array([[2, 0], [0, 2]]) # 放大 2 倍 B = np.array([[0, -1], [1, 0]]) # 旋转 90 度C = A @ B # 先旋转再缩放 x = np.array([1, 0]) print(C @ x) # [0, 2]神经网络中的矩阵 全连接层(Linear 层)的本质就是矩阵乘法加偏置: output = W x input + b一个隐藏层有 128 个神经元、输入维度为 64: import torch import torch.nn as nnlinear = nn.Linear(64, 128) # 内部持有 weight: Tensor (128, 64) 和 bias: Tensor (128,)x = torch.randn(32, 64) # batch_size=32, input_dim=64 y = linear(x) # y.shape = (32, 128)每一层做的都是:把输入向量从一个向量空间线性映射到另一个向量空间。激活函数(ReLU、Sigmoid 等)引入非线性,才让网络能拟合复杂函数。 PyTorch 矩阵运算 import torchA = torch.tensor([[1., 2.], [3., 4.]]) B = torch.tensor([[5., 6.], [7., 8.]])# 矩阵乘法 print(A @ B) # torch.matmul print(torch.mm(A, B)) # 二维矩阵专用# 转置 print(A.T)# 行列式 print(torch.det(A)) # -2.0# 逆矩阵 print(torch.inverse(A))高维 Tensor = 批量矩阵运算 深度学习中输入通常是 3D 以上的 Tensor(batch x dim): # batch matmul:对每个样本独立做矩阵乘法 A = torch.randn(32, 4, 4) # 32 个 4x4 矩阵 B = torch.randn(32, 4, 1) # 32 个列向量 C = torch.bmm(A, B) # (32, 4, 1)torch.bmm 对 batch 中每个矩阵独立相乘,是卷积、Attention 等操作的底层基础。
PyTorch 学习路线:从 Tensor 到训练完整模型
学习阶段规划阶段 时间 目标Python 基础 3~5 天 能写循环、函数、类NumPy 基础 2 天 理解矩阵运算PyTorch Tensor 3 天 会张量计算和 GPU 迁移Autograd 自动求导 2 天 理解梯度和 backwardnn.Module 基础 3 天 会定义 MLP训练流程 3 天 会完整训练和评估CNN 5 天 图片分类Transformer 基础 7 天 理解 Attention项目实战 长期 YOLO、PointNet 等约一个月可以达到能读懂主流论文代码的水平。 为什么从线性回归开始 # 目标:给出面积和房间数,预测房价 area = [80, 100, 120, 150] rooms = [2, 3, 3, 4] price = [120, 180, 220, 300]神经网络就是找一个函数 f,使得 f(area, rooms) ≈ price。PyTorch 的作用是:定义 f 的结构(model) 自动计算 f 的梯度(autograd) 按梯度更新参数,让预测更准确(optimizer)核心训练循环 import torch import torch.nn as nn# 1. 数据 x = torch.tensor([[80, 2], [100, 3], [120, 3], [150, 4]], dtype=torch.float32) y = torch.tensor([120, 180, 220, 300], dtype=torch.float32)# 2. 定义模型 model = nn.Sequential( nn.Linear(2, 16), nn.ReLU(), nn.Linear(16, 1) )# 3. 定义 loss 和 optimizer criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)# 4. 训练循环 for epoch in range(200): pred = model(x).squeeze() # 前向传播 loss = criterion(pred, y) # 计算 loss optimizer.zero_grad() # 清空上一步的梯度 loss.backward() # 反向传播,计算梯度 optimizer.step() # 按梯度更新参数 if epoch % 50 == 0: print(f"Epoch {epoch}: loss = {loss.item():.2f}")四步缺一不可:zero_grad() → 清除旧梯度(PyTorch 默认累积梯度) backward() → 计算所有参数对 loss 的偏导数 step() → 把参数沿梯度方向更新一步Tensor 基础操作 import torcha = torch.tensor([1.0, 2.0, 3.0]) b = torch.zeros(3, 4) c = torch.randn(2, 3) # 标准正态分布# 形状操作 c.shape # torch.Size([2, 3]) c.reshape(3, 2) c.unsqueeze(0) # 增加维度: [1, 2, 3] c.squeeze() # 去掉维度为 1 的维# GPU device = "cuda" if torch.cuda.is_available() else "cpu" c = c.to(device)常用 Loss 函数任务 Loss 调用回归 MSELoss nn.MSELoss()二分类 BCELoss nn.BCEWithLogitsLoss()多分类 CrossEntropy nn.CrossEntropyLoss()常用 Optimizer # Adam:自适应学习率,大多数情况首选 optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)# AdamW:Adam + 权重衰减,Transformer 常用 optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-2)# SGD:经典,配合 momentum optimizer = torch.optim.SGD(model.parameters(), lr=0.01, momentum=0.9)保存和加载模型 # 保存权重 torch.save(model.state_dict(), "model.pth")# 加载权重 model.load_state_dict(torch.load("model.pth")) model.eval() # 推理时关闭 dropout 和 BatchNorm 的训练行为入门建议用 model.state_dict() 而不是直接 torch.save(model, ...),后者会把整个模型类序列化,跨环境加载容易出问题。
PyTorch 点云分割:PointNet 到 Point Transformer 的模型选型
三种任务类型任务 输入 输出 典型场景语义分割 点云 每个点的类别 地面/树木/汽车/行人实例分割 点云 每个点的实例 ID 区分两辆不同车辆部件分割 单个物体点云 零件归属 椅子的腿/靠背/坐垫主流模型 PointNet(最经典) 输入 N×3,直接用 MLP 逐点提取特征,全局池化后再输出分割结果: import torch import torch.nn as nnclass PointNet(nn.Module): def __init__(self, num_classes): super().__init__() self.mlp = nn.Sequential( nn.Linear(3, 64), nn.ReLU(), nn.Linear(64, 128), nn.ReLU(), nn.Linear(128, 256) ) self.cls = nn.Linear(256, num_classes) def forward(self, x): # x: [B, N, 3] feature = self.mlp(x) # [B, N, 256] out = self.cls(feature) # [B, N, num_classes] return out预测时取 argmax 得到每个点的标签: pred = out.argmax(dim=-1) # [B, N]其他模型选型模型 核心 适合场景PointNet++ 局部邻域分层学习 室内、激光雷达、工业检测DGCNN KNN 建图 + EdgeConv 精度要求高的分类/分割Point Transformer Transformer Attention SOTA,大型场景MinkowskiNet 稀疏体素 + Sparse CNN 超密点云,速度快数据格式 点云数据通常有三种格式: xyz # 坐标 [x, y, z] xyzrgb # 坐标 + 颜色 [x, y, z, r, g, b] xyz+intensity # 激光雷达:坐标 + 反射强度读入后转 Tensor: import numpy as np import torchpoints = np.load("scan.npy") # (N, 3) x = torch.tensor(points, dtype=torch.float32) # [N, 3] x = x.unsqueeze(0) # [1, N, 3] add batch dimLoss 函数 点云分割本质是逐点分类,直接用 CrossEntropyLoss: criterion = nn.CrossEntropyLoss()# pred: [B, num_classes, N] or reshape to [B*N, num_classes] # label: [B, N]pred_flat = pred.view(-1, num_classes) # [B*N, num_classes] label_flat = label.view(-1) # [B*N]loss = criterion(pred_flat, label_flat)常用公开数据集数据集 场景 用途ShapeNet Part 单物体 部件分割S3DIS 室内 语义分割SemanticKITTI 自动驾驶 激光雷达语义分割ScanNet RGB-D 室内 场景理解工程推荐流程 点云采集 ↓ 预处理(去噪、下采样、法向量估计) ↓ PointNet++ / Point Transformer 训练 ↓ 输出分割结果 ↓ 后处理(聚类、过滤小区域)入门推荐从 PointNet 开始,结构最简单,容易在本地小数据上跑通。确认流程后再换 PointNet++ 提升精度。
PyTorch 训练基础:Tensor、自动求导、训练循环与模型保存
Tensor Tensor 是 PyTorch 的核心数据结构,类似 NumPy 数组但支持 GPU 运算: import torch# 0 维标量 t0 = torch.tensor(3.14)# 1 维向量 t1 = torch.tensor([1.0, 2.0, 3.0])# 2 维矩阵 t2 = torch.zeros(3, 4)# 移到 GPU if torch.cuda.is_available(): t2 = t2.cuda()Dynamic Graph 与 Autograd PyTorch 使用动态计算图(Define-by-Run),每次前向传播都重新构建计算图,便于调试和条件分支。 开启梯度追踪: x = torch.tensor(2.0, requires_grad=True) y = x ** 2 + 3 * xy.backward() # 反向传播 print(x.grad) # dy/dx = 2x + 3 = 7.0标准训练循环 model = MyModel() optimizer = torch.optim.Adam(model.parameters(), lr=0.001) loss_fn = torch.nn.CrossEntropyLoss()for epoch in range(100): for x_batch, y_batch in dataloader: pred = model(x_batch) # 1. 前向传播 loss = loss_fn(pred, y_batch) # 2. 计算损失 optimizer.zero_grad() # 3. 清空上一步梯度 loss.backward() # 4. 反向传播 optimizer.step() # 5. 更新参数 print(f"Epoch {epoch}, Loss: {loss.item():.4f}")zero_grad() 必须在 backward() 前调用,否则梯度会累加。 损失函数选择任务 损失函数 激活函数二分类 BCELoss Sigmoid多分类 CrossEntropyLoss 无(内置 Softmax)回归 MSELoss 无# 二分类:Logistic Regression model = torch.nn.Sequential( torch.nn.Linear(10, 1), torch.nn.Sigmoid() ) loss_fn = torch.nn.BCELoss()# 多分类 model = torch.nn.Sequential( torch.nn.Linear(10, 5) # 5 个类别 ) loss_fn = torch.nn.CrossEntropyLoss() # 内部含 Softmax# 回归 model = torch.nn.Linear(10, 1) loss_fn = torch.nn.MSELoss()优化器 # SGD(基础,需手动调 lr) optimizer = torch.optim.SGD(model.parameters(), lr=0.01)# SGD + Momentum(更稳定) optimizer = torch.optim.SGD(model.parameters(), lr=0.01, momentum=0.9)# Adam(最常用,lr=0.001 通常无需调整) optimizer = torch.optim.Adam(model.parameters(), lr=0.001)# AdamW(Transformer 推荐,带权重衰减修正) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=0.01)大多数情况从 Adam + lr=0.001 开始,效果不好再调。 保存与加载模型 推荐:只保存参数(state_dict) # 保存 torch.save(model.state_dict(), "model.pth")# 加载 model = MyModel() model.load_state_dict(torch.load("model.pth")) model.eval() # 切换到推理模式保存完整模型(不推荐,依赖类定义路径): torch.save(model, "model_full.pth") model = torch.load("model_full.pth")导出 ONNX ONNX 格式可在 TensorFlow、ONNX Runtime、TensorRT 等框架中使用: dummy_input = torch.randn(1, 3, 224, 224) # 与实际输入形状一致torch.onnx.export( model, dummy_input, "model.onnx", input_names=["input"], output_names=["output"], opset_version=17 )导出后可用 onnxruntime 推理,速度通常比 PyTorch 原生快。 推理模式 model.eval()with torch.no_grad(): # 禁用梯度计算,节省内存 output = model(input_tensor)推理时必须调 model.eval() 和 torch.no_grad(),否则 BatchNorm 和 Dropout 行为与训练时不同。
BIP39 助记词生成原理与 Python 实现
BIP39 定义了一套将随机数编码为可记忆单词的标准,大多数钱包(MetaMask、Phantom、imToken)都支持。 BIP39 原理生成 128/160/192/224/256 位随机熵 取熵的 SHA-256 哈希前 len/32 位作为校验位,拼接到熵末尾 每 11 位映射一个 BIP39 词表单词(共 2048 个词) 最终得到 12/15/18/21/24 个单词熵长度 校验位 总位数 单词数128 bit 4 bit 132 bit 12256 bit 8 bit 264 bit 24Python 实现(bip-utils) pip install bip-utils从助记词派生以太坊私钥和地址 from bip_utils import ( Bip39SeedGenerator, Bip39MnemonicValidator, Bip44, Bip44Coins, Bip44Changes, )mnemonic = "word1 word2 word3 word4 word5 word6 word7 word8 word9 word10 word11 word12"# 可选:验证助记词是否合法 if not Bip39MnemonicValidator().IsValid(mnemonic): raise ValueError("助记词无效")# 助记词 → 种子(可选 passphrase) seed = Bip39SeedGenerator(mnemonic).Generate(passphrase="")# 种子 → BIP44 派生路径 m/44'/60'/0'/0/0 ctx = Bip44.FromSeed(seed, Bip44Coins.ETHEREUM) addr_ctx = ( ctx.Purpose() .Coin() .Account(0) .Change(Bip44Changes.CHAIN_EXT) .AddressIndex(0) )private_key_hex = addr_ctx.PrivateKey().Raw().ToHex() address = addr_ctx.PublicKey().ToAddress()print(f"私钥: {private_key_hex}") print(f"地址: {address}")派生多个地址 for i in range(5): addr_ctx = ( ctx.Purpose() .Coin() .Account(0) .Change(Bip44Changes.CHAIN_EXT) .AddressIndex(i) ) print(f"[{i}] {addr_ctx.PublicKey().ToAddress()}")Solana(BIP44 Coin 501) from bip_utils import Bip44Coinsctx = Bip44.FromSeed(seed, Bip44Coins.SOLANA) sol_ctx = ctx.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT).AddressIndex(0) print(f"Solana 地址: {sol_ctx.PublicKey().ToAddress()}")手动验证 BIP39 校验和(纯 Python) import hashlibdef validate_bip39(mnemonic: str, wordlist_path: str) -> bool: with open(wordlist_path) as f: words = f.read().splitlines() word_index = {w: i for i, w in enumerate(words)} mnemonic_words = mnemonic.strip().split() bits = "" for word in mnemonic_words: if word not in word_index: return False bits += format(word_index[word], "011b") # 最后 len/33 位是校验位 cs_len = len(bits) // 33 entropy_bits = bits[:-cs_len] checksum_bits = bits[-cs_len:] # 还原熵字节 entropy = int(entropy_bits, 2).to_bytes(len(entropy_bits) // 8, "big") # SHA256 前 cs_len 位 h = hashlib.sha256(entropy).digest() expected_cs = format(h[0], "08b")[:cs_len] return checksum_bits == expected_cs注意事项助记词等同于钱包全部资产的控制权,不能在任何联网代码中明文存储 生产环境派生私钥只能在离线机器或 HSM 上操作 仅用于工具开发/验证时,建议使用测试网地址,不要导入真实资产
Python 拿 CPU 型号,以及 Windows 注册表里的 CPU 信息
想在 Python 里拿到完整的 CPU 型号——比如 Intel(R) Xeon(R) CPU E5-2680 v4 @ 2.40GHz 或 12th Gen Intel(R) Core(TM) i5-12400。看着简单,跨平台其实要三条路径。 首选:platform.processor() 标准库一行: import platform print(platform.processor())Linux/macOS 通常能给出可读的字符串。Windows 上经常返回空或者只有个 Intel64 Family 6 Model 158 Stepping 10——CPUID 的原始编码,不是人看的。 Windows:WMI 或注册表 方式 1:wmic import subprocessout = subprocess.check_output("wmic cpu get name", shell=True, text=True) name = out.split("\n")[1].strip() print(name) # 12th Gen Intel(R) Core(TM) i5-12400注意 wmic 在 Windows 11 24H2 之后被标注为已弃用(虽然还能用一段时间)。 方式 2:PowerShell out = subprocess.check_output( ["powershell", "-Command", "(Get-CimInstance Win32_Processor).Name"], text=True, ) print(out.strip())这个在新版 Windows 上更稳。 方式 3:读注册表 Windows 启动时会把 CPUID 结果缓存到注册表: HKLM\HARDWARE\DESCRIPTION\System\CentralProcessor\0 ProcessorNameString REG_SZ 12th Gen Intel(R) Core(TM) i5-12400 VendorIdentifier REG_SZ GenuineIntelPython 读: import winregwith winreg.OpenKey( winreg.HKEY_LOCAL_MACHINE, r"HARDWARE\DESCRIPTION\System\CentralProcessor\0", ) as k: name, _ = winreg.QueryValueEx(k, "ProcessorNameString") print(name)三种方式底层都来自 CPUID 指令,注册表算是"启动时缓存好的解析结果",速度最快,也不依赖外部命令。 Linux:/proc/cpuinfo with open("/proc/cpuinfo") as f: for line in f: if line.startswith("model name"): print(line.split(":", 1)[1].strip()) break或者 shell 一行: grep "model name" /proc/cpuinfo | head -1 lscpu | grep "Model name"macOS:sysctl import subprocess out = subprocess.check_output( ["sysctl", "-n", "machdep.cpu.brand_string"], text=True ) print(out.strip()) # Apple M2 Pro / Intel(R) Core(TM) i7-...跨平台通用函数 import os import platform import subprocessdef get_cpu_name() -> str: system = platform.system() if system == "Windows": try: import winreg with winreg.OpenKey( winreg.HKEY_LOCAL_MACHINE, r"HARDWARE\DESCRIPTION\System\CentralProcessor\0", ) as k: name, _ = winreg.QueryValueEx(k, "ProcessorNameString") return name.strip() except Exception: pass try: out = subprocess.check_output( ["powershell", "-Command", "(Get-CimInstance Win32_Processor).Name"], text=True, ) return out.strip() except Exception: return platform.processor() or "unknown" if system == "Darwin": try: return subprocess.check_output( ["sysctl", "-n", "machdep.cpu.brand_string"], text=True, ).strip() except Exception: return platform.processor() or "unknown" # Linux / *BSD try: with open("/proc/cpuinfo") as f: for line in f: if line.startswith("model name"): return line.split(":", 1)[1].strip() except Exception: pass return platform.processor() or "unknown"if __name__ == "__main__": print(get_cpu_name())CPU 信息会被缓存吗 有个常见疑问:"CPU 型号会不会被系统缓存到什么地方?" 答案分两层:CPU 本身没有"存型号信息"——型号信息是 CPUID 指令的返回值,来自硬件里的固化电路 操作系统会缓存 CPUID 的解析结果:Windows 存在 HKLM\HARDWARE\DESCRIPTION\System\CentralProcessor 注册表下、Linux 通过 /proc/cpuinfo 动态生成、macOS 通过 sysctl 暴露所以你读注册表 / /proc/cpuinfo 拿的是"系统解析过一次的缓存",实时性和 CPUID 指令等价,但在虚拟机 / 容器里可能被覆盖或伪造——CPU 型号做设备指纹是很弱的信号,容易被伪装。 一句话总结 跨平台就 Windows 读注册表、Linux 读 /proc/cpuinfo、macOS sysctl。platform.processor() 在 Linux/macOS 上够用,Windows 上不行。做指纹别只靠 CPU 型号,太容易伪装。
Python 模拟键盘输入:中文乱码和 Citrix 场景的应对
自动化脚本里想模拟键盘输入——听着简单,一遇到中文、遇到 Citrix / RDP / 游戏窗口,坑就一个接一个来。列一下常用方案和各自适用场景。 方案对比库 优点 缺点pyautogui 跨平台、简单 中文经常挂、后台窗口不响应keyboard 全局事件、监听热键 Windows 常需管理员、后台无效pywin32 支持后台 WM_CHAR 某些程序(浏览器、游戏)不响应 WM_CHARctypes + WinAPI 无依赖、最底层 需要处理 scan code、代码稍长剪贴板 + Ctrl+V 中文、长文本最稳 会覆盖用户剪贴板最简单:pyautogui import pyautogui, time time.sleep(3) # 3 秒切窗口 pyautogui.write("hello world") # 输入英文 pyautogui.press("enter") # 按键 pyautogui.hotkey("ctrl", "c") # 组合键interval 参数控制打字间隔: pyautogui.write("hello", interval=0.05)中文乱码怎么办 pyautogui.write("你好") 本质是把字符串拆成一个个按键事件。中文没有对应的物理按键,很多环境(尤其 Citrix、RDP、游戏、浏览器某些输入框)会输入乱码或干脆吞掉。 最稳的做法是 走系统剪贴板 + Ctrl+V: import pyperclip, pyautogui, timepyperclip.copy("你好世界,长文本也行") time.sleep(2) # 切窗口 pyautogui.hotkey("ctrl", "v")优点:支持中文 / emoji / 特殊字符 支持超长文本(比逐字模拟快百倍) 兼容大多数远程桌面 / Citrix缺点:会覆盖用户当前剪贴板(可以备份再还原) 需要目标窗口支持 Ctrl+V带备份还原的完整版本: import pyperclip, pyautogui, timedef paste_text(text): backup = pyperclip.paste() pyperclip.copy(text) time.sleep(0.1) pyautogui.hotkey("ctrl", "v") time.sleep(0.1) pyperclip.copy(backup)后台发送到指定窗口 前台切换太打扰人,想在不抢焦点的情况下往某个窗口发内容: import win32gui, win32api, win32conhwnd = win32gui.FindWindow(None, "记事本") for ch in "hello": win32api.SendMessage(hwnd, win32con.WM_CHAR, ord(ch), 0)局限:浏览器 / 游戏 / Citrix / 远程桌面通常不响应 WM_CHAR(安全考虑) 只对纯 Win32 原生窗口有效不响应的目标只能回到"抢焦点 + 真实按键"路线。 Citrix 里传大数据 场景:本机 → Citrix → 内网服务器,想把一个几 MB 的文件传过去,只能靠键盘输入。 技巧:本机 base64 编码文件(比 hex 省一半体积) 分块用 pyautogui.write 发(每块 4 KB 左右) 内网服务器上再解码import pyautogui, base64, timewith open("payload.bin", "rb") as f: data = base64.b64encode(f.read()).decode()time.sleep(3) # 切到 Citrix 窗口 CHUNK = 4096 for i in range(0, len(data), CHUNK): pyautogui.write(data[i:i + CHUNK], interval=0.001) time.sleep(0.1) pyautogui.press("enter")内网服务器上有 Python / Bash 就 base64 -d 还原。 监听全局热键 想按 F8 触发某个动作: import keyboard keyboard.wait("f8") print("触发")Windows 上需要管理员运行才能读到全局键盘事件。 一句话总结 英文短文本用 pyautogui.write;中文和长文本用剪贴板 + Ctrl+V;Citrix/RDP 传大数据用 base64 + 分块 write。后台无焦点发送只对纯 Win32 窗口有效。
Python 实现 PBKDF2 + AES-256-GCM 加密存储
PBKDF2 + AES-GCM 是本地加密存储(钱包 vault、配置文件保护)的标准方案:密码经 PBKDF2 派生出密钥,AES-GCM 提供加密和完整性验证。 加密流程 Password ↓ PBKDF2-HMAC-SHA512(iterations=600000, salt=随机16字节) ↓ 32 字节 AES Key ↓ AES-256-GCM(nonce=随机16字节) ↓ Ciphertext + Tag安装依赖 pip install pycryptodome完整实现 import json import os import base64 from hashlib import pbkdf2_hmac from Crypto.Cipher import AESPBKDF2_ITERATIONS = 600_000 # NIST 2023 推荐最低值def derive_key(password: str, salt: bytes) -> bytes: return pbkdf2_hmac( "sha512", password.encode("utf-8"), salt, PBKDF2_ITERATIONS, dklen=32, # AES-256 )def encrypt(password: str, plaintext: str) -> dict: salt = os.urandom(16) nonce = os.urandom(16) key = derive_key(password, salt) cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) ciphertext, tag = cipher.encrypt_and_digest(plaintext.encode("utf-8")) return { "salt": base64.b64encode(salt).decode(), "nonce": base64.b64encode(nonce).decode(), "ciphertext": base64.b64encode(ciphertext).decode(), "tag": base64.b64encode(tag).decode(), }def decrypt(password: str, vault: dict) -> str: salt = base64.b64decode(vault["salt"]) nonce = base64.b64decode(vault["nonce"]) ciphertext = base64.b64decode(vault["ciphertext"]) tag = base64.b64decode(vault["tag"]) key = derive_key(password, salt) cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) plaintext = cipher.decrypt_and_verify(ciphertext, tag) return plaintext.decode("utf-8")使用示例 # 加密 secret = '{"privateKey": "0xabcd..."}' vault = encrypt("my_strong_password", secret)# 存储为 JSON with open("vault.json", "w") as f: json.dump(vault, f, indent=2)# 解密 with open("vault.json") as f: vault = json.load(f)plaintext = decrypt("my_strong_password", vault) print(plaintext)vault.json 格式: { "salt": "base64...", "nonce": "base64...", "ciphertext": "base64...", "tag": "base64..." }为什么选 AES-GCM 而非 CBC模式 加密 完整性验证 推荐AES-CBC ✅ ❌(需额外 HMAC) 不推荐新项目AES-GCM ✅ ✅(内置 Tag) ✅ 推���AES-CTR ✅ ❌ 需配合 HMACGCM 模式同时提供加密和认证,decrypt_and_verify 会在解密前验证 Tag,密文被篡改时抛出 ValueError。 关键参数说明 PBKDF2 迭代次数:越高越安全,NIST 2023 建议 SHA-512 至少 210,000 次,MetaMask 等钱包用 600,000 次。代价是派生速度变慢(约 0.5~2 秒),对正常用户不感知,但让暴力破解成本提高数十万倍。 salt 唯一性:每次加密生成新的随机 salt,防止相同密码产生相同密钥(彩虹表攻击)。salt 不需要保密,公开存储即可。 nonce 唯一性:GCM 的 nonce 绝对不能重用于同一密钥,否则 GCM 的安全性完全崩溃。每次加密随机生成是最安全的做法。 密码错误时的行为 try: plaintext = decrypt("wrong_password", vault) except ValueError: print("密码错误或数据被篡改")decrypt_and_verify 验证 Tag 失败时抛出 ValueError,不会泄漏任何明文信息。
Python 读取 YAML:safe_load 和转对象访问
Python 里读 YAML 配置文件,标准做法就是 PyYAML。 安装 pip install pyyaml示例配置 config.yaml: server: host: 127.0.0.1 port: 8000database: user: root password: 123456names: - alice - bob基础读取 import yamlwith open("config.yaml", "r", encoding="utf-8") as f: data = yaml.safe_load(f)print(data["server"]["host"]) # 127.0.0.1 print(data["names"]) # ['alice', 'bob']永远用 safe_load,别用 yaml.load。safe_load 拒绝任意 Python 对象反序列化,避开加载不受信配置时的 RCE 风险。 转对象访问(少写引号) 字典下标写多了嫌烦,可以套一层 SimpleNamespace: from types import SimpleNamespace import yamlwith open("config.yaml", encoding="utf-8") as f: data = yaml.safe_load(f)cfg = SimpleNamespace(**data) print(cfg.server) # 但 cfg.server 还是 dict只包一层不够——嵌套的字典还得递归转: from types import SimpleNamespace import yamldef to_obj(d): if isinstance(d, dict): return SimpleNamespace(**{k: to_obj(v) for k, v in d.items()}) if isinstance(d, list): return [to_obj(i) for i in d] return dwith open("config.yaml", encoding="utf-8") as f: cfg = to_obj(yaml.safe_load(f))print(cfg.server.host) # 127.0.0.1 print(cfg.database.user) # root工程项目更推荐用 pydantic.BaseModel 或 dataclass——能顺带做类型校验: from pydantic import BaseModel import yamlclass ServerCfg(BaseModel): host: str port: intclass Config(BaseModel): server: ServerCfg names: list[str]with open("config.yaml", encoding="utf-8") as f: cfg = Config(**yaml.safe_load(f))print(cfg.server.host)输出 YAML 反向写文件同样常用: import yamldata = {"server": {"host": "0.0.0.0", "port": 80}}with open("out.yaml", "w", encoding="utf-8") as f: yaml.safe_dump(data, f, allow_unicode=True, sort_keys=False)allow_unicode=True 中文不会变成 \uXXXX sort_keys=False 保持字段顺序一句话总结 yaml.safe_load(open(..., encoding="utf-8")) 起步,需要结构化访问就 pydantic。写回用 safe_dump + allow_unicode=True。
Python 读写 YAML:pyyaml safe_load / dump 常用选项
Python 读写 YAML 文件用 PyYAML,核心只有两个函数:yaml.safe_load 和 yaml.dump。 安装 pip install pyyaml读取 YAML # config.yaml server: host: 127.0.0.1 port: 8000database: user: root password: secret123names: - alice - bobimport yamlwith open("config.yaml", "r", encoding="utf-8") as f: data = yaml.safe_load(f)print(data["server"]["host"]) # 127.0.0.1 print(data["names"]) # ['alice', 'bob']用 safe_load 而不是 load——load 会执行 YAML 里嵌入的 Python 代码,存在安全风险。 写出 YAML import yamldata = { "server": {"host": "127.0.0.1", "port": 8000}, "database": {"user": "root", "password": "secret123"}, "names": ["alice", "bob"] }with open("output.yaml", "w", encoding="utf-8") as f: yaml.dump( data, f, allow_unicode=True, # 中文不转成 \uXXXX sort_keys=False # 保持原字典顺序 )默认 sort_keys=True,会按字母排序;allow_unicode=False 会把中文转成 \uXXXX——一般都要加这两个参数。 输出为字符串 yaml_str = yaml.dump(data, allow_unicode=True, sort_keys=False) print(yaml_str)修改 YAML 文件(读取后更新) import yamlwith open("config.yaml", encoding="utf-8") as f: cfg = yaml.safe_load(f)cfg["server"]["port"] = 9000 cfg["names"].append("carol")with open("config.yaml", "w", encoding="utf-8") as f: yaml.dump(cfg, f, allow_unicode=True, sort_keys=False)转成对象访问(点号语法) 如果不喜欢 data["server"]["host"],可以转成 SimpleNamespace: from types import SimpleNamespace import yamldef to_obj(d): if isinstance(d, dict): return SimpleNamespace(**{k: to_obj(v) for k, v in d.items()}) if isinstance(d, list): return [to_obj(i) for i in d] return dwith open("config.yaml", encoding="utf-8") as f: config = to_obj(yaml.safe_load(f))print(config.server.host) # 127.0.0.1 print(config.database.user) # root常见坑 名称含冒号的字符串要加引号,否则 YAML 解析出错: # 错误 name: O:# 正确 name: "O:"字段名问题:Clash Meta 等工具要求字段名是 port,不能是 server_port。读取 sing-box 格式(用 server_port)再写入 Clash 配置时记得手动映射: clash_node = { "name": node["tag"], "server": node["server"], "port": node["server_port"], # 注意字段名 "type": node["type"], }直接 yaml.dump(singbox_node) 生成的字段名 Clash 不认,会报 has unset fields: port。
ModuleNotFoundError: 'Crypto' — pycryptodome 与 pycryptodomex 的坑
代码里: from Crypto.Cipher import AES一跑就崩: ModuleNotFoundError: No module named 'Crypto'网上一查装个 pycrypto——别装,那个包 2013 年就停维护了,还有已知漏洞。 两个包,两个命名空间 现在的正主是这两个,装错任何一个都不通用:包名 导入命名空间pycryptodome from Crypto...pycryptodomex from Cryptodome...pycryptodome:兼容老代码,占用 Crypto 顶层名字 pycryptodomex:改名版本,把命名空间挪到 Cryptodome,避免和老 pycrypto 冲突同一个环境里最多装一个。要是老代码用 from Crypto...,就装 pycryptodome。 装到虚拟环境里,别装到系统 Python PyCharm / VSCode 打开项目通常自动激活了虚拟环境(比如 .venv),但你在别的终端 pip install 时可能装到系统 Python 去了: pip install pycryptodome # 装哪儿了?看运气最稳的写法是显式用当前项目的 Python: D:\Projects\myapp\.venv\Scripts\python.exe -m pip install pycryptodome或者: py -m pip install pycryptodome # Windows py launcher python -m pip install pycryptodome # 当前激活的解释器检查装到哪里了 where python python -c "import sys; print(sys.executable)" python -m pip show pycryptodomeLocation 字段就是包所在目录,肉眼比对一下是不是当前项目的 site-packages。 老代码遇到 pycryptodomex 某些镜像里预装的是 pycryptodomex。老代码不想改的话,兼容层写一下: try: from Crypto.Cipher import AES except ImportError: from Cryptodome.Cipher import AES一句话总结 Crypto = pycryptodome,Cryptodome = pycryptodomex,二选一装到当前项目 Python 里。用 python -m pip install 比裸 pip install 稳。
Python pycryptodome、shutil 目录同步与 tqdm 进度条、PyInstaller 打包
pycryptodome:No module named 'Crypto' 安装了 pycryptodome 但运行时报错: ModuleNotFoundError: No module named 'Crypto'原因:pycryptodome 安装后的导入路径是 Crypto(大写C),和旧版 pycrypto 同名,但两者不兼容。虚拟环境里如果同时装了两个就会冲突。 解决: pip uninstall pycrypto pycryptodome pip install pycryptodome导入方式不变: from Crypto.Cipher import AES from Crypto.Util.Padding import pad, unpad如果想用 Cryptodome 命名空间(与 pycrypto 完全隔离),改装 pycryptodomex: pip install pycryptodomexfrom Cryptodome.Cipher import AES虚拟环境里确认安装: pip list | grep -i crypto # pycryptodome 3.20.0shutil 目录同步加 tqdm 进度条 shutil.copytree 不支持进度回调,需要手动遍历: import shutil from pathlib import Path from tqdm import tqdmdef sync_dir(src: Path, dst: Path): files = [f for f in src.rglob("*") if f.is_file()] dst.mkdir(parents=True, exist_ok=True) for src_file in tqdm(files, desc="复制中", unit="file"): rel = src_file.relative_to(src) dst_file = dst / rel dst_file.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src_file, dst_file)shutil.copy2 会保留原文件的修改时间和权限位。 覆盖前备份,保留最近 N 个 from datetime import datetimedef backup_and_replace(src: Path, dst: Path, keep=5): if dst.exists(): ts = datetime.now().strftime("%Y%m%d_%H%M%S") backup = dst.parent / f"{dst.name}_bak_{ts}" dst.rename(backup) # 清理旧备份,只保留最近 keep 个 baks = sorted(dst.parent.glob(f"{dst.name}_bak_*")) for old in baks[:-keep]: shutil.rmtree(old) sync_dir(src, dst)PyInstaller 打包后路径检测 -F 打包成单文件 exe 后,__file__ 不再可用,需要用 sys.executable: import sys from pathlib import Pathif getattr(sys, "frozen", False): # 打包环境:exe 所在目录 BASE_DIR = Path(sys.executable).parent else: # 开发环境:脚本所在目录 BASE_DIR = Path(__file__).parentCONFIG_PATH = BASE_DIR / "config.json"sys.frozen 在 PyInstaller 打包环境下为 True,普通 Python 解释器下不存在(getattr 返回 False)。 打包命令: pyinstaller -F -n mytool --noconsole main.py-F:单文件 -n mytool:输出名称 --noconsole:不弹出控制台窗口(GUI 工具用)完整示例:插件目录同步工具 import sys import json import shutil from pathlib import Path from datetime import datetime from tqdm import tqdmBASE_DIR = Path(sys.executable).parent if getattr(sys, "frozen", False) else Path(__file__).parent CONFIG = json.loads((BASE_DIR / "config.json").read_text(encoding="utf-8"))SRC = Path(CONFIG["src"]) DST = Path(CONFIG["dst"])def backup(dst: Path, keep=5): if not dst.exists(): return ts = datetime.now().strftime("%Y%m%d_%H%M%S") dst.rename(dst.parent / f"{dst.name}_bak_{ts}") for old in sorted(dst.parent.glob(f"{dst.name}_bak_*"))[:-keep]: shutil.rmtree(old)def copy_with_progress(src: Path, dst: Path): files = [f for f in src.rglob("*") if f.is_file()] dst.mkdir(parents=True, exist_ok=True) for f in tqdm(files, unit="file"): rel = f.relative_to(src) out = dst / rel out.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(f, out)backup(DST) copy_with_progress(SRC, DST) print("同步完成")config.json: { "src": "C:/Users/用户名/AppData/Roaming/adspower_global/cwd_global/chrome/Default", "dst": "D:/backup/chrome_profile" }
Chrome 扩展数据存放位置:LevelDB 与 IndexedDB
Chrome 扩展的 chrome.storage.local 数据并不存在普通的 localStorage 里,而是存在用户数据目录下以 LevelDB 格式保存的独立数据库中。 数据存放路径 Windows # Chrome C:\Users\用户名\AppData\Local\Google\Chrome\User Data\Default\Local Extension Settings\# Edge C:\Users\用户名\AppData\Local\Microsoft\Edge\User Data\Default\Local Extension Settings\每个扩展一个子目录,目录名是扩展 ID: Local Extension Settings\ ├── nkbihfbeogaeaoehlefnkodbefgpgknn\ ← MetaMask │ ├── LOG │ ├── CURRENT │ ├── MANIFEST-000001 │ └── 000003.ldb └── cjpalhdlnbpafiamejdnhcphjbkeiagm\ ← uBlock Origin里面是 LevelDB 格式,不能直接用文本编辑器读取。 macOS ~/Library/Application Support/Google/Chrome/Default/Local Extension Settings/找到扩展 ID 打开 chrome://extensions/,开启右上角开发者模式,即可看到每个扩展的 ID(32 位小写字母)。 在 DevTools 中读取扩展数据 打开扩展的背景页(Service Worker):chrome://extensions/ → 点击扩展的"检查视图:Service Worker"或"background page" 在 DevTools Console 中:// 读取所有 storage.local 数据 chrome.storage.local.get(null, console.log);// 或 const data = await chrome.storage.local.get(); console.log(data);其他存储类型路径 Default\ ├── Local Extension Settings\ ← chrome.storage.local(LevelDB) ├── Extension State\ ← chrome.storage.session ├── IndexedDB\ ← IndexedDB │ └── chrome-extension_xxx.indexeddb.leveldb └── Session Storage\ ← sessionStorageAdsPower 多配置文件批量同步扩展数据 AdsPower 的每个浏览器配置文件存在各自的 cache 目录下,批量同步扩展 LevelDB 数据: import json import shutil import sys from pathlib import Path from datetime import datetime from tqdm import tqdmCONFIG_FILE = Path("config.json") EXT_DIR = "mcohilncbfahbmgdjkbpemcciiolgcge" # 扩展 IDDEFAULT_CONFIG = { "src": r"D:\.ADSPOWER_GLOBAL\cache", "dst": r"D:\ads\.ADSPOWER_GLOBAL\cache" }if not CONFIG_FILE.exists(): with open(CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(DEFAULT_CONFIG, f, indent=4, ensure_ascii=False) print("已创建 config.json,请修改后重新运行") sys.exit(0)with open(CONFIG_FILE, "r", encoding="utf-8") as f: config = json.load(f)src_root = Path(config["src"]) dst_root = Path(config["dst"])# 找出所有包含目标扩展数据的配置文件目录 tasks = [ d for d in src_root.iterdir() if d.is_dir() and (d / "Default" / "Local Extension Settings" / EXT_DIR).exists() ]print(f"发现 {len(tasks)} 个配置需要同步")for index, cache_dir in enumerate(tasks, 1): src = cache_dir / "Default" / "Local Extension Settings" / EXT_DIR dst = dst_root / cache_dir.name / "Default" / "Local Extension Settings" / EXT_DIR print(f"\n[{index}/{len(tasks)}] {cache_dir.name}") # 备份旧目录(加时间戳,不直接删除) if dst.exists(): backup = dst.parent / f"{dst.name}_backup_{datetime.now():%Y%m%d_%H%M%S}" dst.rename(backup) # 只保留最近 5 个备份 backups = sorted(dst.parent.glob(f"{dst.name}_backup_*"), key=lambda p: p.stat().st_mtime, reverse=True) for old in backups[5:]: shutil.rmtree(old) dst.mkdir(parents=True, exist_ok=True) files = [f for f in src.rglob("*") if f.is_file()] with tqdm(total=len(files), desc=cache_dir.name, unit="file", ncols=100) as pbar: for file in files: target = dst / file.relative_to(src) target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(file, target) pbar.update(1)print("\n全部同步完成")关键点:shutil.rmtree 太暴力,改用 rename 加时间戳备份 保留最近 5 个备份,自动清理旧版本 tqdm 显示每个目录的文件级复制进度打包成独立 exe: pip install pyinstaller tqdm pyinstaller -F sync_ext.py
uvicorn reload 模式下找不到文件:用 __file__ 解决相对路径问题
uvicorn 加 --reload 参数时,每次代码变更都会重启子进程。子进程在 import 模块时如果代码顶层有相对路径操作,很容易报 FileNotFoundError。 问题复现 # main.py import cv2img = cv2.imread("small.png", 0) # 顶层直接读取uvicorn main:app --reload启动会报: FileNotFoundError: [Errno 2] No such file or directory: 'small.png'原因:uvicorn reload 子进程的 工作目录(cwd) 可能和你预期的不一致,相对路径 "small.png" 找的是 cwd 下的文件而不是脚本旁边的文件。 解决:用 __file__ 构造绝对路径 from pathlib import PathBASE_DIR = Path(__file__).parentimg = cv2.imread(str(BASE_DIR / "small.png"), 0)Path(__file__).parent 是脚本自身所在目录,不受 cwd 影响,任何启动方式都一致。 统一管理项目路径 from pathlib import PathBASE_DIR = Path(__file__).parent MODELS_DIR = BASE_DIR / "models" DATA_DIR = BASE_DIR / "data" CONFIG_FILE = BASE_DIR / "config.yaml"然后全局使用 BASE_DIR / "xxx" 而不是裸字符串路径。 uvicorn 启动方式 字符串形式(支持 reload): uvicorn.run( "main:app", host="0.0.0.0", port=8000, reload=True )对象形式(不支持 reload): uvicorn.run( app, # FastAPI 对象 host="0.0.0.0", port=8000 # reload=True 在这里无效,reload 必须用字符串形式 )reload 模式会 fork 子进程并 import 指定模块,所以 module 名必须是字符串。 GPU 模型服务的 worker 数 如果 FastAPI 服务里加载了 GPU 模型(SAM、YOLO、PyTorch 等),只开 1 个 worker: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1多 worker 会导致每个进程加载一份模型,显存翻倍。CPU 服务才适合 workers = CPU核数 × 2 + 1。 Nginx upstream 与单进程 单进程 FastAPI 的 upstream 可以简化: upstream backend { server 127.0.0.1:18001; }location / { proxy_pass http://backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; }weight、least_conn 在只有一个节点时没有任何效果,max_fails + fail_timeout 仍然生效(健康检测)。
Python list 转 dict 的几种写法
Python 里"把 list 转成 map"通常指把列表转成字典,方便按 key 查询。几种常见场景: 1. 两个列表合并成 dict keys = ["a", "b", "c"] values = [1, 2, 3]d = dict(zip(keys, values)) print(d) # {'a': 1, 'b': 2, 'c': 3}2. 元组列表转 dict data = [("a", 1), ("b", 2), ("c", 3)] d = dict(data) print(d) # {'a': 1, 'b': 2, 'c': 3}3. 对象列表按字段索引(最常用) 把 list[dict] 转成按某个 key 索引的字典,避免每次用 next(x for x in lst if x["id"] == ...) 线性搜索: users = [ {"id": 1, "name": "Tom"}, {"id": 2, "name": "Jack"}, {"id": 3, "name": "Lucy"} ]user_map = {u["id"]: u for u in users}print(user_map[2]) # {'id': 2, 'name': 'Jack'}这是查询密集型代码里的常用优化:构建一次 O(n),之后每次查找 O(1)。 4. 统计频次(Counter) from collections import Counterlst = ["a", "b", "a", "c", "a"] freq = Counter(lst)print(freq) # Counter({'a': 3, 'b': 1, 'c': 1}) print(freq["a"]) # 3 print(freq.most_common(2)) # [('a', 3), ('b', 1)]Counter 继承自 dict,支持 +、- 合并,常用于词频统计。 5. map() 函数式处理 Python 的 map() 不是字典,而是将函数应用到每个元素: nums = [1, 2, 3, 4]# 不推荐(用 list comprehension 更清晰) result = list(map(lambda x: x * 2, nums))# 推荐 result = [x * 2 for x in nums] print(result) # [2, 4, 6, 8]map() 返回迭代器,需要 list() 转换。列表推导式可读性更好,PyCharm 等工具也不会提示替换。 6. 分组 dict 按某个字段分组: from collections import defaultdictrecords = [ {"type": "A", "val": 1}, {"type": "B", "val": 2}, {"type": "A", "val": 3}, ]grouped = defaultdict(list) for r in records: grouped[r["type"]].append(r)print(dict(grouped)) # {'A': [{'type': 'A', 'val': 1}, {'type': 'A', 'val': 3}], # 'B': [{'type': 'B', 'val': 2}]}快速对比场景 写法两个列表合并 dict(zip(keys, vals))元组列表 dict(data)按字段索引 {item["id"]: item for item in lst}统计频次 Counter(lst)函数处理 [f(x) for x in lst](推荐 comprehension)分组 defaultdict(list)
Python list 转 dict/map 的五种常见姿势
Python 里说"list 转 map"通常指的是把列表变成字典(dict)。到底怎么转,取决于原始 list 长什么样。 1. 两个平行 list 转 dict keys 和 values 是分开的两列: keys = ["a", "b", "c"] values = [1, 2, 3]m = dict(zip(keys, values)) # {'a': 1, 'b': 2, 'c': 3}2. 二元组 list 转 dict 已经是 (key, value) 对: data = [("a", 1), ("b", 2), ("c", 3)] m = dict(data) # {'a': 1, 'b': 2, 'c': 3}3. 字典 list 按字段索引(最常用) 后端返回一堆对象,前端要按 ID 查——最典型场景: users = [ {"id": 1, "name": "Tom"}, {"id": 2, "name": "Jack"}, {"id": 3, "name": "Lucy"}, ]m = {u["id"]: u for u in users}m[2] # {'id': 2, 'name': 'Jack'}如果 key 会重复,改成分组: from collections import defaultdictgroups = defaultdict(list) for u in users: groups[u["dept"]].append(u)4. Python 内建 map() 函数 如果说的是 map() 而不是"映射": nums = [1, 2, 3, 4] result = list(map(lambda x: x * 2, nums)) # [2, 4, 6, 8]更 Pythonic 的写法是列表推导: result = [x * 2 for x in nums]5. 统计次数 用 Counter: from collections import Counterlst = ["a", "b", "a", "c", "a"] Counter(lst) # Counter({'a': 3, 'b': 1, 'c': 1})记忆表输入形态 用法两列 keys + values dict(zip(keys, values))[(k, v), ...] dict(data)[{...}, ...] 按字段 {u[key]: u for u in list}对每个元素求值 [f(x) for x in list]统计次数 Counter(list)
技术分析入门:Stochastic 随机指标与 ATR 波动率
技术分析中,Stochastic(随机指标)用于判断超买超卖,ATR 用于量化波动幅度,两者经常配合使用。 Stochastic 随机指标 %K 线(快线)计算公式: %K = (当前收盘价 - N 周期最低价) / (N 周期最高价 - N 周期最低价) × 100%D 线(慢线)= %K 的移动平均线(通常 3 周期 SMA) 常见参数:Stoch(14, 3, 3) 超买超卖的正确理解Stoch > 80:超买区——价格接近近期高点,不等于马上下跌 Stoch < 20:超卖区——价格接近近期低点,不等于马上反弹强趋势中,Stoch 可以在超买区(> 80)持续数周。直接在超买区做空是新手最常见的错误,强上涨趋势中会被持续轧空。 更可靠的使用方式 震荡行情:超买出现死叉(%K 下穿 %D)→ 考虑空;超卖出现金叉(%K 上穿 %D)→ 考虑多。 趋势行情(顺势交易):下跌趋势中,等反弹到高位出现 Stoch 死叉再做空 上涨趋势中,等回调到低位出现 Stoch 金叉再做多背离信号 顶背离(看跌):价格创新高,但 Stoch 没有创新高——上涨动能减弱。 价格:100 → 110(新高) Stoch:95 → 85(未创新高)背离不代表立即反转,需配合:Stoch 死叉 + 阻力位 + 成交量萎缩,多条件共振才更可靠。 底背离(看涨):价格创新低,但 Stoch 未创新低——空头力量衰竭。 ATR(Average True Range,平均真实波幅) ATR 回答的是:价格最近平均波动有多大,而不是方向。 True Range(TR)计算 每根 K 线的真实波幅取三个值中的最大值: TR = max( H - L, # 本根K线高低差 |H - 昨收|, # 处理向上跳空 |L - 昨收| # 处理向下跳空 )H = 最高价,L = 最低价 ATR(14) 计算 第一个值:前 14 根 K 线 TR 的简单平均。 后续更新(Wilder 平滑法): ATR(t) = ATR(t-1) + (TR(t) - ATR(t-1)) / 14ATR 的实战应用 动态止损: # 不使用固定点数,而是基于当前波动率 stop_loss = entry_price + 1.5 * atr # 做空的止损 take_profit = entry_price - 3 * atr # 做空的止盈固定止损(如 ±2%)在低波动市场太宽,在高波动市场太窄。ATR 止损随市场自适应。 判断追单风险:如果当根 K 线已经运动超过 2~3 个 ATR,追单的风险/收益比明显恶化。 顺势做空的基本逻辑 顺势做空不是"跌多了就空",而是:确认下跌趋势:均线空头排列,价格持续创新低 等反弹:不追空,等价格反弹到阻力位(云带/均线) 找信号:Stoch 在高位死叉,或出现顶背离 进场:止损放在反弹高点上方 1~1.5 ATR需要避免的情况:在强上涨趋势中因为超买做空 在重要支撑位附近追空(风险大,利润空间小) 重大消息前开仓(ATR 在消息前后会剧烈变化,技术指标失效) 小周期信号对抗大周期趋势Python 计算 ATR(简单实现) import pandas as pddef calculate_atr(df, period=14): df["tr"] = pd.concat([ df["high"] - df["low"], (df["high"] - df["close"].shift(1)).abs(), (df["low"] - df["close"].shift(1)).abs(), ], axis=1).max(axis=1) df["atr"] = df["tr"].ewm(alpha=1/period, adjust=False).mean() return dfpandas 的 ewm 实现了类 Wilder 的指数加权平均。
SAM2 / SAM3 文本 Prompt 分割:识别图片中所有特定区域
SAM2 / SAM3 不是问答模型,而是 Promptable Concept Segmentation(PCS) 模型——用自然语言描述一个"概念",模型会找出图中所有匹配的实例,每个都生成一个 mask。 Prompt 写法原则 SAM3 理解的是名词短语(concept),不理解逻辑指令: ✅ 有效 question exam question question block text region❌ 无效 all questions detect all questions find every question on this page写"具体描述词",让模型自己找所有实例。 常用场景 Prompt 试卷 / 文档题目识别 # 基础 text_prompt = "question"# 更精准 text_prompt = "exam question" text_prompt = "question block" text_prompt = "problem statement"负例过滤(排除干扰区域) positive_prompt = "exam question text region" negative_prompt = "header, title, footer, logo, watermark"交通标志识别 text_prompt = "traffic sign" text_prompt = "road sign"物体实例分割 text_prompt = "car" text_prompt = "pedestrian" text_prompt = "building window"Python 调用示例(SAM2 + Grounding DINO 方案) 目前实现 text-prompted 分割的常见管线是 Grounding DINO + SAM2(SAM 官方的 text prompt 能力仍在完善中): from groundingdino.util.inference import load_model, load_image, predict from segment_anything import SAMPredictor# Grounding DINO 定位 boxes, logits, phrases = predict( model=gd_model, image=image, caption="exam question block", box_threshold=0.35, text_threshold=0.25 )# SAM2 分割 predictor.set_image(image) masks, scores, _ = predictor.predict( point_coords=None, point_labels=None, box=boxes, # 把 DINO 的 bbox 传给 SAM multimask_output=False )只做区域切割 + OCR 的方案 SAM 负责"切块",OCR 负责"读内容": # Step 1: SAM 找到所有"题目区域"的 mask masks = sam_segment(image, text_prompt="question block")# Step 2: 按 mask 裁剪图片区域 for i, mask in enumerate(masks): bbox = mask_to_bbox(mask) region = image[bbox[1]:bbox[3], bbox[0]:bbox[2]] # Step 3: OCR 识别文字 text = paddleocr.ocr(region) print(f"题目 {i}: {text}")OCR 推荐:PaddleOCR(中文强)、TrOCR(表格/印刷体强)、Tesseract(通用)。 SAM 的能力边界能力 SAM 支持找出所有"题目"区域 ✅生成 mask/bbox ✅理解"这是第几题" ❌读出题目内容 ❌(需要 OCR)判断题目难度 ❌(需要 LLM)SAM 只做像素级分割,不做语义理解。完整的文档结构化提取需要 SAM + OCR + LLM 组合。 一句话总结 SAM3 文本 prompt 用具体名词短语,它会自动找出所有匹配的实例。复杂场景可以加 negative prompt 排除干扰,或者组合 Grounding DINO + SAM2 + OCR 构建完整的图像理解管线。
SAM3 Prompt 工程:识别文档中的题目区域与多实例分割
SAM3 的核心机制:PCS SAM3 不是问答模型,而是 Promptable Concept Segmentation(PCS) 模型:输入:概念描述(名词短语) 输出:所有匹配该概念的实例 Mask(多个)理解这一点才能写出有效 prompt。 识别题目区域的 Prompt 通用(推荐): exam question question block problem statement更细粒度: question text region numbered question multiple choice question fill-in-the-blank question文档/OCR 类: text question area exercise item不要写 all questions 或 detect all questions,SAM3 不理解 all、detect 这类逻辑词。用名词短语描述"这类东西是什么",SAM3 会自动找出所有实例。 Python 调用示例 import torch from sam3 import SAM3model = SAM3.from_pretrained("facebook/sam3-base") model.eval()from PIL import Image image = Image.open("exam_page.jpg")# PCS 模式:用概念 prompt 找所有实例 with torch.no_grad(): masks = model.predict_concept( image=image, concept="exam question", threshold=0.5 )print(f"找到 {len(masks)} 个题目区域")for i, mask in enumerate(masks): # mask 是 numpy bool array,shape = (H, W) print(f"题目 {i+1}: bbox = {mask_to_bbox(mask)}")与 GroundingDINO 组合使用 SAM3 PCS 直接输出 Mask,但复杂图片中 Mask 边界可能不够精确。可以先用 GroundingDINO 获得检测框,再用 SAM3 精确分割: from groundingdino.util.inference import predict as gdino_predict from sam3 import SAM3Predictor# Step 1: GroundingDINO 获得题目框 boxes, _, _ = gdino_predict( model=gdino_model, image=image, caption="exam question", box_threshold=0.3, text_threshold=0.25 )# Step 2: SAM3 对每个框精确分割 sam3_predictor = SAM3Predictor(model) sam3_predictor.set_image(image)for box in boxes: masks, scores, _ = sam3_predictor.predict( box=box, multimask_output=False ) # masks[0] 即为精确 Mask组合方案比单独使用任一模型效果更好。 Prompt 调试技巧从通用开始,再细化:先试 question,如果噪声太多再改 exam question block同义词扩展:英文 question、problem、exercise、item 可能效果不同,逐一测试调整阈值:threshold 降低会找到更多实例(包括噪声),提高则更保守多 prompt 投票:对同一图片用多个 prompt 预测,取交集或并集prompts = ["exam question", "question block", "problem statement"] all_masks = []for p in prompts: masks = model.predict_concept(image=image, concept=p, threshold=0.4) all_masks.extend(masks)# NMS 去重(基于 IoU) final_masks = nms_masks(all_masks, iou_threshold=0.5)常见场景 Prompt 参考场景 推荐 Prompt试卷题目 exam question, question block表单字段 form field, input field车辆检测 car, vehicle, truck人脸 human face, person face交通标志 traffic sign, road sign文字区域 text region, printed textSAM3 的训练数据以英文概念为主,中文 prompt(如"题目")效果不稳定,建议始终使用英文。
Python 多线程检查 HTTP 状态码并删除 403 文件
批量检查 data/*.json 里的图片 URL,删除所有返回 403 的 JSON 文件。 单线程版本 import json from pathlib import Pathimport requestsfor json_file in Path("data").glob("*.json"): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: continue url = data[0].get("imgHttpUrl") if not url: continue r = requests.get(url, stream=True, timeout=10) if r.status_code == 403: print(f"删除 {json_file}") json_file.unlink() except Exception as e: print(json_file, e)只检查数组第一个元素的 URL,够快。若一个文件里所有 URL 都要检查,把 data[0] 改成循环即可。 多线程版本(100 并发) import json from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completedimport requestsDATA_DIR = Path("data") MAX_WORKERS = 100session = requests.Session()def check_file(json_file: Path): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: return url = data[0].get("imgHttpUrl") if not url: return # 优先用 HEAD,减少带宽 r = session.head(url, timeout=10, allow_redirects=True) # 部分服务器不支持 HEAD,降级到 GET if r.status_code == 405: r = session.get(url, stream=True, timeout=10) if r.status_code == 403: print(f"[删除] {json_file}") json_file.unlink() except Exception as e: print(f"[错误] {json_file}: {e}")files = list(DATA_DIR.glob("*.json"))with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = [executor.submit(check_file, f) for f in files] for _ in as_completed(futures): passprint("完成")requests.Session 在多线程中是线程安全的,复用 TCP 连接比每次创建新 session 快。 并发数选择文件数 建议 workers< 1000 501000~10000 100> 10000 考虑 aiohttp + asyncioHTTP 请求是 IO 密集型,100 线程一般没问题。若服务端有限流,调小 MAX_WORKERS。 aiohttp 异步版(超大规模) 文件数超过 10 万时,aiohttp 比线程池快几倍: import asyncio import json from pathlib import Pathimport aiohttp import aiofilesDATA_DIR = Path("data") CONCURRENCY = 500async def check_file(session, json_file: Path): try: async with aiofiles.open(json_file, encoding="utf-8") as f: content = await f.read() data = json.loads(content) if not data: return url = data[0].get("imgHttpUrl") if not url: return async with session.head(url, allow_redirects=True, timeout=aiohttp.ClientTimeout(total=10)) as r: if r.status == 403: print(f"[删除] {json_file}") json_file.unlink() except Exception as e: print(f"[错误] {json_file}: {e}")async def main(): files = list(DATA_DIR.glob("*.json")) sem = asyncio.Semaphore(CONCURRENCY) async def bounded(f): async with sem: await check_file(session, f) async with aiohttp.ClientSession() as session: await asyncio.gather(*(bounded(f) for f in files))asyncio.run(main())按 answer 目录过滤删除(补充场景) 如果是 downloads/ 里没有对应 answer/ 文件就删除,可以改为本地比对: from pathlib import Pathdownloads_dir = Path("downloads") answer_dir = Path("answer")# answer 目录中所有文件名(不带后缀) answer_stems = {f.stem for f in answer_dir.iterdir() if f.is_file()}for file in downloads_dir.iterdir(): if not file.is_file(): continue if file.stem not in answer_stems: print("删除", file) file.unlink()不需要网络请求,纯本地文件对比,速度极快。
Python 多线程批量检查 URL 状态码:ThreadPoolExecutor + Session
一个真实场景:data/*.json,每个 JSON 是数组,数组元素里有 imgHttpUrl。想批量检查这些 URL 是不是 403(图片失效),是的话把整个 JSON 文件删掉。文件几千个,单线程要跑一天,加多线程。 单线程版本先跑通 import json from pathlib import Path import requestsfor json_file in Path("data").glob("*.json"): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: continue url = data[0].get("imgHttpUrl") if not url: continue r = requests.get(url, stream=True, timeout=10) if r.status_code == 403: print(f"删除 {json_file}") json_file.unlink() except Exception as e: print(json_file, e)要点:只检查数组第一个元素(假设整个文件的图片来自同一源,一个 403 全体 403) stream=True — 只拿响应头,body 不下载,快 用 pathlib.Path.unlink() 删文件,比 os.remove 干净加多线程 IO 密集,直接 ThreadPoolExecutor 就行,不用协程: import json from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import requestsDATA_DIR = Path("data") MAX_WORKERS = 100session = requests.Session()def check_file(json_file: Path): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: return None url = data[0].get("imgHttpUrl") if not url: return None r = session.get(url, stream=True, timeout=10, allow_redirects=True) if r.status_code == 403: json_file.unlink() return json_file except Exception as e: print(f"[错误] {json_file}: {e}") return Nonefiles = list(DATA_DIR.glob("*.json")) deleted = []with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: futures = {ex.submit(check_file, f): f for f in files} for fut in as_completed(futures): result = fut.result() if result: deleted.append(result)print(f"完成,共删除 {len(deleted)} 个文件")Session 为什么关键 用 requests.Session() 而不是每次 requests.get():复用 TCP 连接 + TLS 握手,快 5~10 倍 连接池自动管理 全局默认 headers / cookies批量请求同一个域名,Session 是必配的。 HEAD 还是 GET? 只判状态码,不需要 body,理论上 HEAD 更快: r = session.head(url, timeout=10, allow_redirects=True)但:CDN / 图片服务器可能不支持 HEAD,返回 405 Method Not Allowed(常见于阿里 OSS 未开该权限、部分反爬策略) 有些服务器 HEAD 返回和 GET 不同的状态码保险起见先 HEAD、失败降级 GET: def check(url): try: r = session.head(url, timeout=10, allow_redirects=True) if r.status_code == 405: r = session.get(url, stream=True, timeout=10, allow_redirects=True) return r.status_code except requests.RequestException: return None线程数怎么调CPU 密集:os.cpu_count() 一般顶 IO 密集(网络请求):可以开几十甚至几百 目标服务器扛得住多少:看它——同源打 100 并发容易被 WAF ban,不同源随便开试起来:从 20 开始,看服务器响应正常、CPU 也没打满、就慢慢加。100 是 Python requests 常见上限。 更快:httpx + asyncio 千 URL 级别 ThreadPoolExecutor 够用。上万级建议改 asyncio: import asyncio, httpx, json from pathlib import Pathasync def check(client, json_file): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: return url = data[0].get("imgHttpUrl") if not url: return r = await client.head(url, follow_redirects=True, timeout=10) if r.status_code == 403: json_file.unlink() print(f"删除 {json_file}") except Exception as e: print(json_file, e)async def main(): files = list(Path("data").glob("*.json")) limits = httpx.Limits(max_connections=200) async with httpx.AsyncClient(limits=limits) as client: await asyncio.gather(*(check(client, f) for f in files))asyncio.run(main())asyncio + httpx 可以轻松跑到 1000+ 并发,机器压力更小。 一句话总结 IO 密集批量请求 = ThreadPoolExecutor + Session。开 50-100 线程,先 HEAD 再降级 GET。上万量级换 asyncio + httpx。
SAM3 标注交通牌主体:正负点提示策略
用 SAM3 标注交通标志时,默认分割经常会把灯杆、支架、阴影一起圈进来。关键在于提示点的位置和正负样本的配合。 单点提示(最简方案) 把提示点打在牌面中央,SAM 通常只会分割牌面:牌型 提示点位置圆形限速牌 圆心三角警示牌 三角形中间矩形指示牌 牌面中央points = [[x_center, y_center]] labels = [1] # 1 = foreground正点 + 负点过滤杆子 如果单点还是把杆子带入,加负样本点排除: points = [ [牌面中心x, 牌面中心y], # 正样本 [杆子中间x, 杆子中间y], # 负样本 ] labels = [1, 0] # 1=前景, 0=背景实际调用: masks, scores, logits = predictor.predict( point_coords=np.array(points), point_labels=np.array(labels), multimask_output=True, )# 取得分最高的 mask best_mask = masks[np.argmax(scores)]检测框缩小(配合 Grounding DINO) Grounding DINO 给出的 bbox 通常包含了灯杆,传给 SAM 时手动收缩: def shrink_box(box, ratio=0.85): x1, y1, x2, y2 = box cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 w, h = (x2 - x1) * ratio, (y2 - y1) * ratio return [cx - w/2, cy - h/2, cx + w/2, cy + h/2]tight_box = shrink_box(dino_box, ratio=0.8)masks, scores, _ = predictor.predict( point_coords=None, point_labels=None, box=np.array(tight_box), multimask_output=False, )缩小 15~20% 通常能去掉杆子区域。 mask 面积过滤 有时 SAM 返回多个候选 mask,按面积过滤掉太大或太小的: img_area = image.shape[0] * image.shape[1]valid_masks = [] for mask in masks: area = mask.sum() ratio = area / img_area if 0.01 < ratio < 0.4: # 占图片面积 1%~40% valid_masks.append(mask)交通牌通常不超过画面的 40%,过滤掉背景误判。 完整管线(Grounding DINO + SAM3) from groundingdino.util.inference import load_model, predict from segment_anything import SamPredictor# Step 1: DINO 定位交通牌 boxes, logits, phrases = predict( model=gd_model, image=image, caption="traffic sign", box_threshold=0.35, text_threshold=0.25, )# Step 2: 收缩框 tight_boxes = [shrink_box(b) for b in boxes]# Step 3: SAM 精确分割 predictor.set_image(image) for box in tight_boxes: masks, scores, _ = predictor.predict( box=np.array(box), multimask_output=False, ) # 保存 mask...调参建议 杆子还是出现 → 加负样本点在杆子上,或再缩小 bbox 5% 牌面被切掉 → bbox 缩小太多,调回 ratio=0.9 误检到路面标线 → 提高 DINO 的 box_threshold(0.35 → 0.45) 多块牌聚在一起 → multimask_output=True 后手动选面积最合适的 mask
Python 依赖版本冲突:requires-python 约束与 numpy/torch 的 Python 版本限制
报错解读 × No solution found when resolving dependencies for split (markers: python_full_version == '3.8.*'): Because numpy>=1.26.0 depends on Python>=3.9,<3.13, we can conclude that numpy>=1.26.0 cannot be used on Python 3.8.这不是 numpy 装不上,而是 requires-python 声明的范围与依赖的 Python 要求相矛盾:项目声明支持 Python>=3.8 numpy>=1.26 要求 Python>=3.9,<3.13 求解器为 Python 3.8 找不到满足条件的 numpy 版本,报 No solution found修复方案 收窄 requires-python 范围(推荐): # pyproject.toml [project] requires-python = ">=3.9,<3.13" # 或更保险 requires-python = ">=3.10,<3.13"这是现代 ML/CV 依赖(numpy>=1.26、torch>=2、OpenCV)能稳定运行的范围。 如果必须保留 Python 3.8(不推荐): numpy==1.24.4 # 最后支持 Python 3.8 的 numpy 版本但 torch>=2、transformers、sam 等几乎都不再支持 3.8,维护成本极高。 Python 版本与主要 ML 库支持范围库 最低 Python 最高 Pythonnumpy >= 1.26 3.9 3.12torch >= 2.0 3.8 3.12torch >= 2.3 3.9 3.12transformers >= 4.40 3.8 3.12统一建议:Python 3.10 或 3.11,兼容范围最广,且有长期支持。 triton 只支持 Linux + CUDA × No solution found when resolving dependencies: Because only triton==0.4.1...2.1.0 are available and triton has no wheels for Windowstriton 是 GPU kernel 编译库,官方只发布 Linux + CUDA 的 wheel,在 Windows 上无法通过 pip 安装。 处理方式: # 将 triton 设为可选依赖,或仅在 Linux 下安装 [project.optional-dependencies] gpu = ["triton>=2.0; sys_platform == 'linux'"]或者在 Windows 开发环境使用 WSL2 / Docker。 uv 常用排查命令 # 查看当前 Python 版本 uv python list# 指定 Python 版本安装 uv pip install numpy --python 3.11# 查看依赖树 uv pip show numpy# 锁定依赖版本 uv pip compile pyproject.toml -o requirements.txt
uv 依赖冲突:requires-python 太宽 + Windows 装 triton 的双坑
上一个 CV 项目,uv sync 直接崩: × No solution found when resolving dependencies for split (markers: python_full_version == '3.8.*'): ↳ Because the requested Python version (>=3.8) does not satisfy Python>=3.9.0 and numpy>=1.26.0 depends on Python>=3.9,<3.13, we can conclude that numpy>=1.26.0 cannot be used.看起来是"numpy 装不上",其实是 Python 版本范围和依赖范围打架。 症状识别:requires-python 太宽 pyproject.toml 里如果写了: requires-python = ">=3.8"uv 会尝试为 3.8 / 3.9 / ... / 3.12 全部找一套解。numpy ≥ 1.26 要求 Python ≥ 3.9,Python 3.8 直接被排除——solver 认为"你说要支持 3.8,但 3.8 没解",结论:无解。 解法:收紧 requires-python 改成现代 AI 项目实际支持的范围: requires-python = ">=3.10,<3.13"理由:numpy ≥ 1.26 要求 Python ≥ 3.9 torch 2.x + Windows 的 wheel 覆盖到 3.12 3.13 太新,很多 native wheel 还没出之后 uv sync 通常就过了。 另一坑:Windows 上 triton 无解 同一个项目继续装: uv pip install triton× No solution found: triton<=2.1.0 has no wheels with a matching Python ABI tag (e.g., cp312) triton>=2.2.0 has no wheels with a matching platform tag (win_amd64)不是版本冲突,是 triton 官方根本不出 Windows wheel。Linux 有 manylinux_x86_64,macOS 部分版本有,Windows 是空白。 Windows 上处理 triton 的三条路 方案 A:上 WSL2(推荐) 装个 Ubuntu,在里面: python3.11 -m venv venv source venv/bin/activate pip install triton torchCUDA、torch、triton 生态齐活。 方案 B:直接放弃 triton 只是跑 inference 或 demo,很多模型不强依赖 triton: uv remove triton代码里如果有 import triton,包一层 try/except 或者按平台条件装: [project.optional-dependencies] gpu = ["triton; sys_platform == 'linux'"]方案 C:走 Linux 服务器 有云 GPU 就直接扔上去,从头就没 Windows 这个问题。 uv 常用的调试思路 # 看清依赖树里谁在拉某个包 uv tree | grep numpy# 只解析不装,快速试冲突 uv lock# 指定用某个 Python uv sync --python 3.11# 显式约束某个包 uv add "numpy>=1.26,<2"一句话总结 uv 报 no solution,先看两点:requires-python 范围是不是包了依赖不支持的旧 Python、目标平台是不是根本没轮子。Windows 上装 triton 是无解,别死磕。
Python 通过 Shadowsocks SOCKS5 代理发送请求
核心原理 Python 不能直接解析 Shadowsocks 协议。标准做法是: Python ↓ SOCKS5 sslocal (本地 1080 端口) ↓ Shadowsocks 协议 远端 SS 服务器 ↓ 明文 目标网站先用 Shadowsocks 客户端在本地起一个 SOCKS5 代理,Python 通过这个代理访问外网。 方法一:requests[socks](推荐) 安装: pip install requests[socks]使用: import requestsproxies = { "http": "socks5://127.0.0.1:1080", "https": "socks5://127.0.0.1:1080" }r = requests.get( "https://httpbin.org/ip", proxies=proxies, timeout=10 )print(r.json())socks5:// 使用远端 DNS 解析;如果想让本地 DNS 解析,换成 socks5h://。 方法二:PySocks 全局劫持 安装: pip install pysocks设置全局默认代理,之后所有 socket 连接都走 SOCKS5: import socket import sockssocks.set_default_proxy(socks.SOCKS5, "127.0.0.1", 1080) socket.socket = socks.socksocketimport requestsr = requests.get("https://httpbin.org/ip") print(r.json())适合需要代理所有网络调用(requests、httpx、urllib 等)的场景,但会影响进程内所有 socket,慎用。 方法三:环境变量 export ALL_PROXY=socks5://127.0.0.1:1080 export HTTPS_PROXY=socks5://127.0.0.1:1080 export HTTP_PROXY=socks5://127.0.0.1:1080或者在 Python 代码里设置: import osos.environ["ALL_PROXY"] = "socks5://127.0.0.1:1080"import requestsr = requests.get("https://httpbin.org/ip") print(r.json())requests 会自动读取 HTTP_PROXY / HTTPS_PROXY / ALL_PROXY 环境变量。 在代码里启动 sslocal 如果想让 Python 程序自己管理 Shadowsocks 子进程: import subprocess import time import requestsproc = subprocess.Popen([ "sslocal", "-s", "服务器IP", "-p", "8388", "-k", "密码", "-m", "aes-256-gcm", "-l", "1080" ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)time.sleep(1) # 等待 sslocal 就绪proxies = { "http": "socks5://127.0.0.1:1080", "https": "socks5://127.0.0.1:1080" }try: r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10) print(r.json()) finally: proc.terminate()sslocal 来自 shadowsocks-libev 或 shadowsocks-rust 包,需要提前安装。 验证代理是否生效 import requestsproxies = { "http": "socks5://127.0.0.1:1080", "https": "socks5://127.0.0.1:1080" }r = requests.get("https://httpbin.org/ip", proxies=proxies, timeout=10) print(r.json()) # 返回的 origin IP 应为代理服务器 IP,不是本机 IP
目标检测辅助标注:GroundingDINO、SAM2 与 Florence-2 实用指南
为什么需要辅助标注 手动标注几万张图片耗时数周,且容易出现漏标和错标。开放词汇检测(Open-Vocabulary Detection)模型可以:输入文字描述的类别 直接输出检测框 人工只做最终校验流程: 原始图片 ↓ GroundingDINO(文字→检测框) ↓ 人工检查 + 修正 ↓ YOLO 训练GroundingDINO:文字驱动检测 不需要预先训练,直接用类别名称描述来检测: from groundingdino.util.inference import load_model, load_image, predictmodel = load_model("groundingdino_swint_ogc.pth", "config.py") image_source, image = load_image("image.jpg")boxes, logits, phrases = predict( model=model, image=image, caption="car . truck . person . traffic light", box_threshold=0.35, text_threshold=0.25 )caption 用英文句点分隔类别,box_threshold 控制检测置信度。返回的 boxes 是归一化的 (cx, cy, w, h) 格式,可直接转为 YOLO 标签。 3060Ti 8G 可运行:GroundingDINO-SwinT(轻量,速度快) GroundingDINO-SwinB(更准,显存要求更高)SAM2:检测框转 Mask SAM2(Segment Anything Model 2)在 GroundingDINO 输出的框基础上生成精确分割轮廓: from sam2.build_sam import build_sam2 from sam2.sam2_image_predictor import SAM2ImagePredictorsam2 = build_sam2("sam2_hiera_small.pt", device="cuda") predictor = SAM2ImagePredictor(sam2)predictor.set_image(image_np)# 用 GroundingDINO 的框作为 prompt masks, scores, _ = predictor.predict( box=boxes_xyxy, # [x1, y1, x2, y2] multimask_output=False )组合效果: GroundingDINO(框)+ SAM2(Mask)= 实例分割标注适合标注车辆、行人等需要精确轮廓的场景。 Florence-2:一键生成多种格式 微软的 Florence-2 是视觉语言模型,支持多种任务: from transformers import AutoProcessor, AutoModelForCausalLMmodel = AutoModelForCausalLM.from_pretrained("microsoft/Florence-2-large") processor = AutoProcessor.from_pretrained("microsoft/Florence-2-large")# 开放词汇检测 inputs = processor( text="<OPEN_VOCABULARY_DETECTION>car", images=image, return_tensors="pt" ) outputs = model.generate(**inputs) result = processor.decode(outputs[0], skip_special_tokens=False)支持的任务标签:<OD> — 通用目标检测 <CAPTION> — 图像描述 <OPEN_VOCABULARY_DETECTION>类别 — 指定类别检测 <OCR> — 文字识别Qwen2.5-VL:标注质量审核 GroundingDINO 标注后,用 Qwen2.5-VL 批量审核: # 伪代码示例 for image, label in dataset: response = qwen_vl.chat( image=image, question="图中的标注框是否正确?是否有漏标?" ) if "错误" in response or "漏标" in response: flag_for_review(image)批量过滤低质量标注,减少人工检查量。 推荐工具:X-AnyLabeling X-AnyLabeling 是集成了上述模型的图形化标注工具: pip install x-anylabeling x-anylabeling内置 GroundingDINO、SAM2、YOLO 等模型,支持:点击即分割 批量自动预标注 导出 YOLO、COCO、VOC 格式标注流程建议 1. 用 GroundingDINO 批量预标注(自动化) 2. X-AnyLabeling 打开,批量审查明显错误 3. SAM2 对需要 Mask 的目标细化边界 4. Qwen2.5-VL 辅助审核复杂场景 5. 导出 YOLO 格式,开始训练第一轮训出小模型后,用模型自身的预测结果辅助后续标注,形成正向循环(主动学习)。
RTX 3060 Ti 能跑哪些 AI 模型:LLM、OCR、YOLO 全景
RTX 3060 Ti(8GB 显存)是个甜点卡,覆盖了 LLM 推理、YOLO 训练、OCR、图像生成等主要本地 AI 场景。 LLM 大语言模型 流畅运行(推荐) 量化后直接跑,速度可接受:模型 4bit 显存Qwen3-4B 2~3 GBQwen3-8B(4bit) 4~6 GBLlama 3.2 3B 2~3 GBGemma 3 4B 2~3 GB勉强可用Qwen3-14B(需要部分层卸载到 CPU) Llama 3.1 8B(长上下文时速度明显下降)不推荐Qwen3-32B、DeepSeek-R1 原版、70B 级模型——需要 24GB+ 显存或多卡。工具推荐:Ollama(最简单)、LM Studio(有 GUI)、vLLM(高吞吐推理)。 OCR工具 适用场景PaddleOCR / PP-OCRv5 中文、表格、票据,CPU 也能跑Surya 复杂文档,精度高EasyOCR 部署最简单Qwen2.5-VL-7B 复杂界面截图,可输出结构化 JSON对于包含布局和表格的文档,PaddleX 的 PP-StructureV3 可以直接输出 HTML 或 Excel。 YOLO + OCR 组合是工业自动化常见方案: 截图 → YOLO 找目标框 → PaddleOCR 读文字 → 业务逻辑这比直接上 VL 大模型快几十倍。 YOLO 训练型号 状态YOLOv8n / YOLO11n 完全没问题YOLOv8s / YOLO11s 完全没问题YOLOv8m / YOLO11m 可以,batch 调小YOLOv8l / YOLOv8x 较吃力,batch=1~2 或降图片尺寸自动标注:GroundingDINO + SAM2 GroundingSAM2 是最强的自动标注组合: 文字提示("car", "person") → GroundingDINO 找框 → SAM2 生成分割轮廓 → 自动生成标注文件优点:开放词汇检测,不需要重新训练类别。 各模型比较:模型 支持任意类别 自动标注质量GroundingSAM2 ✅ ⭐⭐⭐⭐⭐⭐GroundingDINO SwinB-QInt8 ✅ ⭐⭐⭐⭐⭐GroundingDINO SwinT-QInt8 ✅ ⭐⭐⭐⭐DEIMv2-DINOv3(COCO) ❌ 只支持训练类别 ⭐⭐⭐如果数据量很大(几万张),先用 SwinT 批量跑,再人工抽查修正。 多模态 VLMQwen2.5-VL-7B:图片问答、OCR、表格识别、UI 截图分析 Qwen2.5-VL-3B:更轻量,3060Ti 跑很轻松 MiniCPM-V 4.5:轻量多模态用途:图片问答、自动生成标注描述、分析网页截图。 图像生成Stable Diffusion 1.5 / SDXL(低分辨率):512×512 很舒服 FLUX.1-dev:需要量化版,显存刚好够用语音Whisper:视频转字幕,中文效果强 CosyVoice / GPT-SoVITS:语音克隆,几秒样本就能复刻声线IndexedDB 中提取图片 如果标注数据存在浏览器 IndexedDB 里,可以用控制台导出: const req = indexedDB.open("数据库名");req.onsuccess = () => { const db = req.result; const store = db.transaction("Video2DImage", "readonly").objectStore("Video2DImage"); store.getAll().onsuccess = e => { e.target.result.forEach((item, i) => { const blob = item.image || item.data || item.blob; if (!blob) return; const url = URL.createObjectURL(blob); const a = document.createElement("a"); a.href = url; a.download = `${i}.jpg`; a.click(); setTimeout(() => URL.revokeObjectURL(url), 1000); }); }; };数量很多时建议用 JSZip 打包成 zip 后统一下载,避免浏览器拦截大量弹窗。 推荐配置(32GB 内存 + 3060Ti) Qwen3-8B → 代码助手、日常问答 Qwen2.5-VL-7B → 看图、OCR、标注辅助 Whisper large → 语音转文字 GroundingDINO + SAM2 → 自动标注 YOLO11s → 目标检测训练这套组合基本把 8GB 显存榨干,覆盖开发、标注、OCR、视觉分析全场景。
Python from __future__ import annotations 详解:延迟求值与前向引用
它解决什么问题 Python 在解析函数签名时会立即对类型注解求值。如果类还没定义完就引用了自身,会抛 NameError: class Node: def __init__(self, next: Node = None): # NameError: name 'Node' is not defined self.next = next加上这一行就能解决: from __future__ import annotationsclass Node: def __init__(self, next: Node = None): # 正常 self.next = next原理是让 Python 把所有注解保存为字符串,不再立即解析。 核心效果 from __future__ import annotationsclass User: passdef get_user() -> User: return User()print(get_user.__annotations__) # {'return': 'User'} ← 字符串,不是 <class '__main__.User'>没有这行时,__annotations__ 里存的是实际类对象;有了这行,全部变成字符串,运行时不解析。 解决循环引用 from __future__ import annotationsclass A: def test(self, b: B) -> None: passclass B: def test(self, a: A) -> None: pass不加的话,定义 A 时 B 还不存在,报 NameError。 常见使用场景 FastAPI / Pydantic 模型互相引用: from __future__ import annotations from pydantic import BaseModelclass Article: author: Authorclass Author: articles: list[Article]SQLAlchemy ORM 关系: from __future__ import annotations from sqlalchemy.orm import Mapped, relationshipclass User(Base): posts: Mapped[list[Post]] = relationship()class Post(Base): user: Mapped[User] = relationship()这是 FastAPI、Pydantic、SQLAlchemy 等库的文件开头几乎都有这行的原因。 运行时获取实际类型 如果需要在运行时把字符串注解解析回真实类型,用 typing.get_type_hints(): from __future__ import annotations import typingclass Foo: def bar(self) -> Foo: return selfhints = typing.get_type_hints(Foo.bar) print(hints) # {'return': <class '__main__.Foo'>}get_type_hints() 会在运行时把字符串注解解析成真实类型。 Python 版本说明from __future__ import annotations 来自 PEP 563,Python 3.7 引入 PEP 563 原本计划在 Python 3.10 成为默认行为,后来推迟,至今(Python 3.12+)仍需显式导入 Python 3.10 引入了 X | Y 联合类型语法,可以写 int | None 代替 Optional[int],但前向引用问题 from __future__ import annotations 仍是最简洁的解法一行导入,让注解写法更自由,是现代 Python 项目的常见约定。
Python 环境管理工具选谁:uv / conda / poetry / pixi 对比
Python 世界的环境管理工具太多——pip、virtualenv、pipenv、poetry、conda、mamba、rye、uv、pixi、hatch……哪个最适合现在(2026 年)用?简单结论:大多数项目 uv,AI/CUDA 项目考虑 pixi 或 conda。 对比表工具 环境管理 包管理 速度 主要适用 现状uv ✅ ✅ 🚀 最快 大多数 Python 项目 活跃、事实标准pixi ✅ ✅ 🚀 很快 AI/CUDA/科学计算 新兴,快速崛起conda ✅ ✅ 慢 AI、科学计算 稳定但笨重mamba ✅ ✅ 快 conda 生态替代 活跃poetry ✅ ✅ 中 库/应用打包 成熟,用户多pipenv ✅ ✅ 慢 老项目 维护缓慢rye ✅ ✅ 快 曾是首选 已合并入 uvhatch ✅ ✅ 快 多环境测试 小众pip + venv ✅ ✅ 慢 兜底 官方最基础方案uv:默认选它 uv 是 Astral(写 Ruff 的那家)用 Rust 实现的一体化工具,把这些的能力都整合了:pyenv(Python 版本管理) virtualenv(虚拟环境) pip(包管理) pip-tools(依赖锁定) pipx(全局工具)日常操作: uv init # 初始化项目 uv python install 3.11 # 装 Python 3.11 uv add requests # 加依赖 uv add --dev pytest # 开发依赖 uv sync # 同步依赖到 .venv uv run main.py # 用项目环境跑 uv tool install ruff # 全局装工具(类似 pipx)速度上,pip 装 20 个包要 30 秒,uv 通常 2-3 秒。CI 尤其明显。 什么时候不用 uv CUDA 依赖复杂 + 遇到 wheel 找不到: 比如:Windows 上装 triton — 直接没轮子(见 uv 依赖冲突) OpenCV 特殊构建(Contrib、GPU) CUDA 版本要精确匹配 PyTorch Intel MKL / Nvidia cuDNN 直接依赖这些场景conda / pixi 生态里有预编译好的二进制,uv 从 PyPI 拿不到就得自己编译(往往编不过)。 Pixi:AI 项目的新选择 Pixi 由 conda 生态团队做的下一代工具——Rust 实现,速度接近 uv,能力接近 conda: pixi init pixi add python=3.11 pytorch=2.4 pixi add opencv # 自动解决 cuda / mkl pixi run python main.py优势:兼容 conda-forge 生态(大量预编译二进制) 自动解 CUDA / MKL / OpenCV 类非 Python 依赖 支持多语言(Python、Rust、C++、Julia) 速度快,比 conda 快得多已经有相当多 AI 开发者从 conda 迁移到 pixi。 Conda / Mamba:老牌 AI 首选 conda 的最大优势是 conda-forge 里几十万个预编译二进制包——包括 CUDA、cuDNN、MKL、OpenCV、FFmpeg 等 Python 生态之外的东西。 问题:解依赖巨慢(几分钟起步) 环境臃肿(一个 base 就几个 G) 私有仓库通常不支持Mamba 是 conda 的兼容替代,速度快 10 倍,用法一样: mamba install pytorch cudatoolkit=11.8 mamba env create -f environment.yml新项目直接上 pixi,比 conda + mamba 都简单。 Poetry:库开发 写 Python 库、要发 PyPI,poetry 还是很好的选择: poetry init poetry add requests poetry publish优势:强项在依赖锁定 + 发布 wheel pyproject.toml 规范 依赖冲突处理清晰uv 现在也能做发布(uv build + uv publish),但生态积累不如 poetry。 Rye 呢 Rye 之前是 Astral 主推的项目,2024 年宣布功能合并到 uv,Rye 只维护不新增。已经用 Rye 的可以直接迁到 uv,语法几乎一样。 场景推荐场景 首选 备选Web / 脚本 / 日常开发 uv poetry库开发 + 发 PyPI uv poetryYOLO / PyTorch / LLM 训练 uv(先试) pixi / mamba科学计算 + CUDA + MKL pixi mamba老项目改造 保持原有工具 逐步迁 uvCI 加速 uvuv 装全局工具 uv 不太强调"全局环境",但支持全局工具(类似 pipx): uv tool install ruff uv tool install pre-commit uv tool install pyinstalleruv tool list uv tool upgrade ruff uv tool uninstall ruff装完直接命令行可用。想真的用一个全局 Python 解释器写脚本,可以: uv python install 3.11 uv run --python 3.11 script.py一句话总结 默认 uv,CUDA 复杂就上 pixi。conda/mamba 老 AI 项目继续维护,poetry 专注库发布。Rye 已经并入 uv,不用再学。
YOLO 标注工具选型:X-AnyLabeling / LabelMe / PaddleX
YOLO 训练项目中,标注工具的选择影响效率,特别是在自动预标注和多种格式导出方面差异很大。 X-AnyLabeling(推荐,功能最全) 国内团队维护,目前更新最活跃的智能标注工具。 核心优势:内置 SAM 自动分割(点击目标自动生成 mask) 集成 GroundingDINO 文本驱动检测 支持 YOLOv8 预标注(先训一个小模型,用它预标注剩余数据) 支持矩形、多边形、关键点、旋转框等多种标注类型 支持 YOLO、COCO、VOC 格式导出pip install x-anylabeling x-anylabeling适合:需要 AI 辅助加速标注,或做实例分割。 LabelMe(经典稳定) Python 实现,教程资料最多,适合新手入门。 pip install labelme labelme特点:支持矩形、多边形、圆形、线段标注 JSON 格式保存,可转换为 YOLO/COCO 格式 社区资料丰富不足:纯手动标注,没有 AI 辅助,标注效率相对低。 PaddleX(无代码完整流程) 百度飞桨出品,从标注到训练到部署的一站式平台。 pip install paddlepaddle paddlex完整流程: 数据上传 → 标注 → 训练(可视化监控)→ 评估 → 导出模型 → 部署特点:图形化界面,不需要写代码 支持目标检测、图像分类、OCR、语义分割 自带数据增强 支持 GPU 分布式训练适合:不熟悉深度学习代码的工程师,或做快速验证。 Label Studio(团队协作) 功能最完整的企业级标注平台。 pip install label-studio label-studio start特点:多用户任务分配和审核 支持图片、文本、音频、视频 REST API,可集成到训练流水线 支持 YOLO、COCO 等格式导出 社区版免费,企业版收费适合:团队协作标注大数据集,或需要标注审核流程。 工具对比工具 AI 辅助 团队协作 学习成本 适用场景X-AnyLabeling ✅ SAM+DINO ❌ 低 智能辅助标注LabelMe ❌ ❌ 极低 个人小数据集PaddleX 部分 ❌ 低(有 GUI) 无代码训练Label Studio ✅ ✅ 中 团队大规模标注YOLO 格式快速说明 标注完成后导出为 YOLO 格式(.txt 文件): # 每行格式:class_id cx cy width height 0 0.5 0.3 0.2 0.15 1 0.7 0.6 0.1 0.08坐标均为相对图片尺寸的归一化值(0~1)。class_id 从 0 开始,对应 data.yaml 中的类别列表。 预标注加速工作流 效率最高的策略:手动标注 200~500 张 训练一个初版 YOLOv8 模型 用该模型对剩余数据预标注 人工审核+修正(只改错误的框) 合并数据重新训练通常从第二轮开始,实际手工修改量减少 60~80%。
Playwright 接管本地 Chrome:四种方式对比
Playwright 默认下载并使用自带的 Chromium。想用本机装的 Google Chrome(自动化登录时保留 cookie、复用插件),有四种方式,各自适用场景不同。 方式一:channel = "chrome" 最省事——channel: "chrome" 告诉 Playwright 用系统装的稳定版 Chrome: from playwright.sync_api import sync_playwrightwith sync_playwright() as p: browser = p.chromium.launch(channel="chrome", headless=False) page = browser.new_page() page.goto("https://example.com") input("回车退出") browser.close()优点:跨平台一行搞定 局限:不会用你日常那个 profile,不带 cookie、不带插件其它可选 channel:msedge、chrome-beta、chrome-dev、msedge-beta。 方式二:executable_path 指定路径 Chrome 装在非默认路径,或者想用 Chromium 兼容分发(Brave、Edge、Vivaldi): Windows: browser = p.chromium.launch( executable_path=r"C:\Program Files\Google\Chrome\Application\chrome.exe", headless=False, )macOS: browser = p.chromium.launch( executable_path="/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", headless=False, )Linux: browser = p.chromium.launch( executable_path="/usr/bin/google-chrome", headless=False, )跟方式一一样,不带 cookie / 插件——只是换了浏览器可执行文件。 方式三:launchPersistentContext 复用 profile 想让 Playwright 用日常那个 Chrome 用户目录(复用登录态、书签、扩展): browser = p.chromium.launch_persistent_context( user_data_dir=r"C:\Users\<你>\AppData\Local\Google\Chrome\User Data", channel="chrome", headless=False, ) page = browser.pages[0] if browser.pages else browser.new_page() page.goto("https://example.com")注意:这个方式要求 Chrome 没在运行,否则报: Failed to launch: browser closed. Profile is already in use解法:关掉所有 Chrome 窗口再跑 或者复制一份 profile 出来单独用:xcopy "C:\...\User Data" "D:\pw-profile" /E /I想只用某个 Profile(不是 Default): browser = p.chromium.launch_persistent_context( user_data_dir=r"D:\pw-profile", args=["--profile-directory=Profile 1"], channel="chrome", headless=False, )方式四:CDP 连接已启动的 Chrome(推荐) 方式三最大的痛点是"Chrome 必须关闭"。想在 Chrome 正常开着的情况下让 Playwright 接管它,走 CDP(Chrome DevTools Protocol)。 第一步:以 debug 模式启动 Chrome,指定端口: "C:\Program Files\Google\Chrome\Application\chrome.exe" ^ --remote-debugging-port=9222 ^ --user-data-dir="D:\pw-chrome-profile"--user-data-dir 指向一个专用目录,避免和日常 Chrome 抢占用。 第二步:Playwright 连上去: from playwright.sync_api import sync_playwrightwith sync_playwright() as p: browser = p.chromium.connect_over_cdp("http://127.0.0.1:9222") ctx = browser.contexts[0] page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto("https://example.com")优势:Chrome 一直开着,做完自动化 Playwright 断开就行 保留登录态、插件、cookie 可以边脚本操作边人工插手(做半自动化)生产爬虫、账号自动登录场景基本用这个。 四种方式对比方式 Chrome 需关闭 保留 profile 使用场景channel="chrome" 否 否 简单自动化,无需登录态executable_path=... 否 否 用其它 Chromium 分发launch_persistent_context 是 是 一次性用现有 profileconnect_over_cdp (CDP) 否 是 推荐:常驻 Chrome + 脚本联动顺带:无痕、隐身、下载路径 无痕模式: browser = p.chromium.launch(channel="chrome", args=["--incognito"])指定下载目录: context = browser.new_context(accept_downloads=True) page = context.new_page() async with page.expect_download() as info: await page.click("a#download") download = await info.value await download.save_as("D:/downloads/file.zip")忽略证书错误(内网自签证书): context = browser.new_context(ignore_https_errors=True)一句话总结 普通自动化用 channel="chrome"、要复用登录态用 CDP 连接(先手动启动 Chrome + --remote-debugging-port=9222)。launch_persistent_context 只在 Chrome 关闭时才能用。
Playwright 使用本地已安装的 Chrome:channel、executablePath 与 CDP 连接
Playwright 默认使用自带的 Chromium 浏览器。以下四种方式可以改用本地安装的 Google Chrome。 方法一:channel='chrome'(最简单) 需要本机已安装 Google Chrome。 Python: from playwright.sync_api import sync_playwrightwith sync_playwright() as p: browser = p.chromium.launch(channel="chrome", headless=False) page = browser.new_page() page.goto("https://example.com") browser.close()JavaScript: const { chromium } = require('playwright');const browser = await chromium.launch({ channel: 'chrome', headless: false }); const page = await browser.newPage(); await page.goto('https://example.com'); await browser.close();方法二:指定 executablePath Python(各平台路径): # Windows executable_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"# macOS executable_path = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"# Linux executable_path = "/usr/bin/google-chrome"browser = p.chromium.launch(executable_path=executable_path, headless=False)方法三:launchPersistentContext(复用登录状态) 使用已登录账号、Cookie、插件的 Chrome 用户数据目录: Python: from playwright.sync_api import sync_playwrightwith sync_playwright() as p: context = p.chromium.launch_persistent_context( r"C:\Users\用户名\AppData\Local\Google\Chrome\User Data", channel="chrome", headless=False, ) page = context.pages[0] or context.new_page() page.goto("https://example.com")注意:Chrome 运行时会独占用户目录,使用此方法前需要关闭所有 Chrome 窗口,否则报错: Profile is already in use方法四:connectOverCDP(连接已启动的 Chrome) 先启动 Chrome 并开启远程调试端口(不影响正常使用): # Windows "C:\Program Files\Google\Chrome\Application\chrome.exe" --remote-debugging-port=9222# macOS /Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome --remote-debugging-port=9222然后 Playwright 连接: Python: browser = p.chromium.connect_over_cdp("http://127.0.0.1:9222") page = browser.contexts[0].pages[0] page.goto("https://example.com")JavaScript: const browser = await chromium.connectOverCDP('http://127.0.0.1:9222'); const page = browser.contexts()[0].pages()[0];这种方式完全复用当前 Chrome 的登录状态、Cookie 和扩展,且不独占用户目录。适合需要手动操作与自动化混合使用的场景。 方案对比方法 复用登录 Chrome 需关闭 保留插件channel='chrome' ✗ 否 ✗executablePath ✗ 否 ✗launchPersistentContext ✓ 是 ✓connectOverCDP ✓ 否 ✓
量化交易中的 Universe:股票池定义、过滤规则与数据获取
Universe 是什么 量化交易里,Universe(投资宇宙 / 股票池)指策略允许选股和交易的资产集合。简单理解: Universe = 你的策略在哪些股票里选比如:沪深 300 成分股(300 只) 中证 500 成分股(500 只) 全 A 股剔除 ST、停牌、次新股后的剩余标的 某个行业的所有上市公司为什么要定义 Universe A 股有 5000+ 只股票。如果不做筛选,直接在全市场跑因子计算:计算量大,速度慢 流动性差的股票出信号无法成交 次新股、ST 股信号质量差,噪声多先定义好 Universe,再在 Universe 里做因子计算、排名、选股,是量化策略的标准流程: 全市场(5000+只) ↓ Universe 过滤 符合条件的股票池(如 3000 只) ↓ 因子计算 每只股票打分 ↓ 排序选股 前 N 名 → 下单交易静态 Universe 直接用指数成分股,每次调仓时更新成分股列表: import akshare as ak# 沪深 300 成分股 df = ak.index_stock_cons_csindex(symbol="000300") universe = df["成分券代码"].tolist()# 中证 500 df = ak.index_stock_cons_csindex(symbol="000905") universe = df["成分券代码"].tolist()动态 Universe 每天收盘后重新计算,根据流动性、市值等条件动态生成: import pandas as pddef build_universe(daily_data: pd.DataFrame) -> list: """ daily_data: 包含 code, close, volume, market_cap, is_st, list_days 等列 """ universe = daily_data[ (~daily_data["is_st"]) & # 剔除 ST (daily_data["list_days"] >= 60) & # 上市满 60 天 (daily_data["volume"] > 0) & # 非停牌 (daily_data["market_cap"] > 5e9) & # 市值 > 50 亿 (daily_data["turnover_20d"] > 5e7) # 20 日均成交额 > 5000 万 ] return universe["code"].tolist()常用 Universe 类型Universe 说明 适合策略沪深 300 成分股 蓝筹龙头,流动性好 低换手率、稳健因子中证 500 成分股 中盘股,成长性较好 中小盘因子中证 1000 成分股 小盘股,弹性大 小盘动量、题材全 A 剔除 ST/停牌 覆盖面广 全市场扫描行业 Universe 限定某一行业 行业轮动、配对交易在量化框架中使用 JoinQuant(聚宽): from jqdata import get_index_stocksuniverse = get_index_stocks("000300.XSHG") # 沪深 300Qlib: # qlib 配置文件 market: csi300vn.py / 自研框架: universe = build_universe(daily_df)for code in universe: factor_score = compute_factor(code) ...Universe 与 AI Agent 量化 在 AI Agent 驱动的量化系统里,Universe 还有另一层含义:限定 Agent 的分析范围。 universe: - A股 - 港股 - ETF配置好 Universe 后,Agent 的新闻抓取、因子计算、持仓分析都限制在这个范围内,避免无限制扩散导致计算量和 Token 消耗失控。
Python JSONDecodeError 排查:NaN、不完整结构、数字字符串的常见坑
JSONDecodeError 定位 json.loads() 报错时,JSONDecodeError 对象携带精确位置信息: import jsontry: data = json.loads(content) except json.JSONDecodeError as e: print(f"错误信息: {e.msg}") print(f"行号: {e.lineno}, 列号: {e.colno}") print(f"字符偏移: {e.pos}") print(f"出错内容: {content[max(0, e.pos-50):e.pos+50]}")原因一:裸 NaN / Infinity JSON 规范不允许 NaN 和 Infinity 作为数值字面量,只有加了引号的字符串 "NaN" 才合法: # 合法 json.loads('{"value": "NaN"}') # OK,返回字符串# 非法 json.loads('{"value": NaN}') # JSONDecodeError: Expecting value json.loads('{"value": Infinity}') # JSONDecodeError如果数据源返回了裸 NaN,需要预处理: import recleaned = re.sub(r'\bNaN\b', '"NaN"', content) data = json.loads(cleaned)原因二:内容不是完整 JSON 常见错误是将多个 JSON 对象直接拼接(不用数组包裹): # 非法:两个对象并排 content = '{"a": 1}, {"b": 2}' json.loads(content) # JSONDecodeError# 合法:用数组包裹 content = '[{"a": 1}, {"b": 2}]' json.loads(content) # OK另一种情况是只收到了响应的一部分(网络截断),pos 错误位置通常在字符串末尾。 原因三:数字存为字符串 JSON 中数字加了引号,解析后得到字符串而非数值: {"x": "1228.8650628969294", "z": "NaN"}data = json.loads(content) type(data["x"]) # str,不是 float# 需要手动转换 x = float(data["x"])# z 是 "NaN" 字符串,转 float 得到 nan z = float(data["z"]) # float('nan') import math math.isnan(z) # True原因四:编码问题 非 UTF-8 编码的文件用 json.loads 读取会报错: # 读取文件时指定编码 with open('data.json', 'r', encoding='utf-8') as f: data = json.load(f)# 如果来源是 bytes data = json.loads(raw_bytes.decode('utf-8'))快速调试模板 import json, tracebackdef safe_parse(content: str): try: return json.loads(content), None except json.JSONDecodeError as e: context_start = max(0, e.pos - 30) context_end = min(len(content), e.pos + 30) snippet = content[context_start:context_end] marker = ' ' * (e.pos - context_start) + '^' print(f"JSONDecodeError at line {e.lineno}, col {e.colno}:") print(snippet) print(marker) return None, e
Python JSONDecodeError 排查清单:从 NaN 到编码 BOM
json.loads 抛 JSONDecodeError 时,异常信息里有行号、列号、字节偏移量——先把这三个印出来,问题基本就定位了: import jsontry: data = json.loads(content) except json.JSONDecodeError as e: print(e) print("line:", e.lineno, "col:", e.colno, "pos:", e.pos) print(content[max(0, e.pos - 40): e.pos + 40])最后一行把出错位置前后 40 个字符打出来,肉眼就能看出问题。以下是几种常见踩坑。 1. 根本不是合法 JSON(最常见) 抓了一段"对象序列"贴进来: { "id": 1 }, { "id": 2 }这不是合法 JSON,缺个数组外壳。要么改成: [ { "id": 1 }, { "id": 2 } ]要么按字段包一层: { "objects": [...] }2. NaN / Infinity 没引号 标准 JSON 不支持 NaN、Infinity。这样合法: {"z": "NaN"} ← 字符串这样非法: {"z": NaN} ← Python json.loads 会崩服务端如果是 Python json.dumps(..., allow_nan=True) 输出的,可能就是裸 NaN。解决办法: import json data = json.loads(content, parse_constant=lambda x: None) # 或 data = content.replace("NaN", "null") data = json.loads(data)3. 结尾多逗号 也是非标准 JSON: { "a": 1, "b": 2, ← 这个逗号 Python 不接受 }用 json5 或者预处理正则去掉最后一个逗号: import re content = re.sub(r",\s*([}\]])", r"\1", content)4. UTF-8 BOM 从 Windows 记事本另存的 JSON 常带 BOM,第一个字符是 : json.loads(content) # JSONDecodeError: Expecting value: line 1 column 1 (char 0)读文件时用 utf-8-sig: with open("data.json", encoding="utf-8-sig") as f: data = json.load(f)5. 数字全被当字符串了 不会抛异常,但会让下游算错: {"x": "1228.86"}type(obj["x"]) 是 str,要计算得先转: x = float(obj["x"])批量转的话用一个小函数: def to_float_fields(d, fields): for f in fields: if f in d: d[f] = float(d[f]) return d6. 单引号不是 JSON 从 Python print(dict) 复制的字符串是单引号: {'id': 1, 'name': 'Tom'}这不是 JSON,是 Python repr。要么请对方 json.dumps 输出,要么本地 ast.literal_eval: import ast data = ast.literal_eval("{'id': 1, 'name': 'Tom'}")literal_eval 只解析字面量,比 eval 安全得多。 一句话总结 JSONDecodeError 第一步永远是把 e.pos 附近的原文打出来。九成情况能在 30 秒内定位到具体字符是啥。
分页 API 全量遍历:Python 与 JavaScript 实现模式
分页接口通常返回 totalCount、pageSize、当页数据。需要全量数据时,用 while 循环按页请求即可。 Python 实现 按 totalCount 推进(稳定写法) import mathpage_size = 20 current = 1 all_items = []while True: params = { "status": "ON_SALE", "current": current, "pageSize": page_size } response = fetchProductList(params) data = response.get("model", []) all_items.extend(data) total_count = response.get("totalCount", 0) total_pages = math.ceil(total_count / page_size) print(f"已获取第 {current}/{total_pages} 页") if current >= total_pages: break current += 1print(f"总共获取 {len(all_items)} 条")按当页数量判断末页(防 totalCount 不准) 如果接口的 totalCount 有时不准确,用当页返回数量是否小于 pageSize 来判断末页更稳: page_size = 20 current = 1 all_items = []while True: params = {"status": "ON_SALE", "current": current, "pageSize": page_size} response = fetchProductList(params) items = response.get("model", []) if not items: break all_items.extend(items) print(f"获取第 {current} 页,{len(items)} 条") if len(items) < page_size: break current += 1生成器模式(内存友好) 适合商品数量很大、需要边翻页边处理的场景: def iter_products(): current = 1 page_size = 20 while True: resp = fetchProductList({ "status": "ON_SALE", "current": current, "pageSize": page_size }) items = resp.get("model", []) if not items: break yield from items if len(items) < page_size: break current += 1for item in iter_products(): print(item["title"])JavaScript 实现 async/await 版本 async function getAllProducts() { const pageSize = 20; let current = 1; const allItems = []; while (true) { const response = await fetchProductList({ status: "ON_SALE", current, pageSize }); const items = response?.model || []; if (items.length === 0) { break; } allItems.push(...items); console.log(`获取第 ${current} 页,${items.length} 条`); if (items.length < pageSize) { break; } current++; } console.log(`总共获取 ${allItems.length} 条`); return allItems; }const products = await getAllProducts();按 totalCount 计算总页数 async function getAllProducts() { const pageSize = 20; let current = 1; let totalPages = 1; const allItems = []; while (current <= totalPages) { const response = await fetchProductList({ status: "ON_SALE", current, pageSize }); const items = response?.model || []; allItems.push(...items); totalPages = Math.ceil((response.totalCount || 0) / pageSize); console.log(`获取第 ${current}/${totalPages} 页`); current++; } return allItems; }流式处理(不存全量数据) 适合直接对每条数据执行操作(改价格、同步库存等),避免一次性把几千条数据加载到内存: let current = 1; const pageSize = 20;while (true) { const response = await fetchProductList({ status: "ON_SALE", current, pageSize }); const items = response?.model || []; if (!items.length) { break; } for (const item of items) { // 在这里处理每条商品 await updateStock(item.id, item.quantity); } if (items.length < pageSize) { break; } current++; }两种终止条件对比策略 适用场景 风险current >= totalPages totalCount 准确 totalCount 不准时多/少请求items.length < pageSize 更通用 最后一页恰好整除时多一次空请求两种可以结合使用:先按 totalPages 推进,同时检查当页数量作为保险。
YOLO11 预训练模型指南:Ultralytics、Roboflow Universe 与 HuggingFace
Ultralytics 官方预训练模型 YOLO11 由 Ultralytics 维护,提供五档大小(n/s/m/l/x),覆盖四类主要任务:任务 示例权重 用途目标检测 yolo11n.pt ~ yolo11x.pt 输出 bbox + 类别实例分割 yolo11n-seg.pt ~ yolo11x-seg.pt 输出精确 Mask姿态识别 yolo11n-pose.pt ~ yolo11x-pose.pt 人体关键点旋转框(OBB) yolo11n-obb.pt ~ yolo11x-obb.pt 遥感/倾斜目标默认基于 COCO 80 类训练,可直接检测人、车、猫狗、手机、键盘等常见目标: from ultralytics import YOLOmodel = YOLO("yolo11m.pt") results = model("image.jpg")推理返回的 results 包含 bbox 坐标(xywh 或 xyxy)、置信度和类别 ID。 实例分割 model = YOLO("yolo11m-seg.pt") results = model("image.jpg")# 获取多边形点集(可直接转 YOLO 分割格式) for r in results: for mask in r.masks.xy: print(mask) # [[x, y], ...]垂直场景预训练权重 不需要从头训练的常见场景权重可在以下平台获取: Roboflow Universe:提供数千个社区训练好的 YOLO 权重,支持直接下载或 API 调用。常见方向:安全帽检测(工地、工厂) 烟火检测 跌倒检测 车牌识别 PCB 缺陷检测 路面裂缝HuggingFace:搜索 "yolo11" 或 "yolov8" 可找到专有领域微调版本,包含医学图像、农业、无人机航拍等方向。 更多可用模型 除 YOLO11 外,以下模型各有优势:模型 特点 适用场景YOLO-World 文本描述直接检测任意目标 零样本检测YOLO12 Attention 架构,精度更高 精度优先RT-DETR Transformer 检测器,速度快 部署优先RF-DETR 小数据集效果好 数据少的场景YOLOX 无 AGPL 协议争议 商业部署PP-YOLOE 百度飞桨出品 国内生态YOLO-World 示例 from ultralytics import YOLOmodel = YOLO("yolov8s-worldv2.pt")# 设置类别(无需重新训练) model.set_classes(["car", "person", "safety helmet"])results = model("image.jpg")输入类别名称即可检测,适合快速验证和标注辅助。 推荐工作流 没有垂直场景数据时: 1. 从 Roboflow Universe 找现有权重 2. 用预训练权重对自己数据集预标注 3. 人工审核 + 修正 4. 在预训练权重基础上微调(而非从头训练)微调比从头训练所需的数据量少 3~5 倍,收敛速度也更快。
MITM 抓包代理的实现方案:从用户态代理到驱动层
六种实现层次对比方案 典型工具 特点用户态代理 mitmproxy, Charles, Burp Suite 配置简单,需安装根证书TUN 虚拟网卡 Clash Meta, sing-box, Surge 不依赖系统代理,可抓 UDP/游戏WinDivert WinDivert + 自定义程序 Windows 驱动层,游戏加速器常用WFP 企业安全/DLP 产品 微软官方,性能高,不易检测NDIS Filter Wireshark/Npcap 二层帧,可抓所有流量Socket Hook Frida, Xposed 拦截加密前明文,绕过证书绑定用户态代理(最常见) 流量路径: App → 系统代理 → MITM Proxy → 目标服务器HTTPS 需要:生成 CA 根证书,安装到系统信任 收到 CONNECT hostname:443 后,动态签发 hostname 的域名证书 与客户端建立一条 TLS,与服务端建立另一条 TLS,在中间解密/重新加密TUN 虚拟网卡方案 App → 虚拟网卡(TUN) → 用户态协议栈 → 真实网卡Linux 创建 TUN: int tun_fd = open("/dev/net/tun", O_RDWR); // 配置后:所有流量走 TUN,用户态读 IP 包 read(tun_fd, buf, sizeof(buf));优点:不需要配系统代理,UDP 也能抓,Clash/sing-box 都用这套。 Python 实现最小 HTTPS MITM 第一步:生成 CA openssl genrsa -out ca.key 2048 openssl req -x509 -new -key ca.key -out ca.crt -days 3650安装 ca.crt 到系统根证书信任。 第二步:动态签发域名证书 from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa import datetimedef gen_cert(hostname: str, ca_key, ca_cert): key = rsa.generate_private_key(public_exponent=65537, key_size=2048) cert = ( x509.CertificateBuilder() .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)])) .issuer_name(ca_cert.subject) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.datetime.utcnow()) .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365)) .add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname)]), critical=False) .sign(ca_key, hashes.SHA256()) ) return key, cert第三步:TLS 劫持 import ssl# 客户端侧:用伪造证书与客户端握手 def mitm_tls_client(client_socket, hostname, certfile, keyfile): ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) ctx.load_cert_chain(certfile, keyfile) return ctx.wrap_socket(client_socket, server_side=True)# 服务端侧:正常连接真实服务器 def connect_server(hostname, port): ctx = ssl.create_default_context() raw = socket.create_connection((hostname, port)) return ctx.wrap_socket(raw, server_hostname=hostname)两端建立后,中间就能读到明文 HTTP 请求: req = client_tls.recv(8192) print(req.decode()) # POST /v1/chat/completions ...HTTP 解析推荐 不要手动解析 HTTP,使用成熟库: pip install h11 # 轻量,推荐 # 或 pip install httptools证书缓存 对同一域名动态生成一次后缓存,避免每次重复生成: cert_cache = {}def get_cert(hostname): if hostname not in cert_cache: key, cert = gen_cert(hostname, ca_key, ca_cert) cert_cache[hostname] = (key, cert) return cert_cache[hostname]最小模块划分 自己实现一个基础 HTTPS MITM 代理,核心代码约 500~1000 行: proxy/ ├── listener.py # 监听连接 ├── connect.py # 处理 CONNECT 方法 ├── tls.py # TLS 劫持 ├── cert.py # 动态证书生成/缓存 ├── http.py # HTTP 解析 └── logger.py # 请求日志局限性 用户态 HTTP Proxy 在以下场景会遇到问题:QUIC/HTTP3(基于 UDP,不走 CONNECT) App 硬编码不走系统代理 Certificate Pinning(固定证书哈希) gRPC over HTTP2对于需要捕获所有流量的场景(如 AI Agent 全局监听),TUN + gVisor Netstack 扩展性更好。
2026 年 AI Agent 框架选型:LangGraph、Mastra、PydanticAI 对比与 LangGraph 最小实现
框架横向对比(2026)框架 语言 适合场景LangGraph Python 生产环境、多 Agent、长流程Mastra TypeScript Next.js 全栈项目PydanticAI Python FastAPI、结构化输出CrewAI Python 多角色协作 AgentAutoGen Python 研究方向、自动代码执行OpenAI Agents SDK Python 轻量 Tool Calling、OpenAI 为主综合推荐排名:LangGraph > OpenAI Agents SDK > Mastra > PydanticAI > CrewAI > AutoGen LangGraph:生产首选 LangGraph 基于状态机(StateGraph),天然支持:循环图(Think → Act → Observe → Think) Checkpoint 持久化 Multi-Agent(Supervisor / Swarm) Human-in-the-Loop(Interrupt + Resume) 与 Playwright/Browser Use 集成大量生产环境已从 LangChain Agent 迁移至 LangGraph。 LangGraph 最小实现 pip install langgraph langchain-openai langchain-coretools.py: from langchain_core.tools import tool import subprocess@tool def search(query: str) -> str: """搜索信息""" return f"搜索结果: {query}"@tool def shell(command: str) -> str: """执行命令""" try: return subprocess.check_output(command, shell=True, text=True, timeout=5)[:5000] except Exception as e: return str(e)app.py: from typing import TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, ToolMessage from tools import search, shellllm = ChatOpenAI(model="gpt-4o") tools = [search, shell] llm_with_tools = llm.bind_tools(tools)class AgentState(TypedDict): messages: listdef agent_node(state): response = llm_with_tools.invoke(state["messages"]) return {"messages": state["messages"] + [response]}def tool_node(state): last = state["messages"][-1] tool_messages = [] for call in last.tool_calls: result = search.invoke(call["args"]) if call["name"] == "search" else shell.invoke(call["args"]) tool_messages.append(ToolMessage(content=str(result), tool_call_id=call["id"])) return {"messages": state["messages"] + tool_messages}def should_continue(state): last = state["messages"][-1] return "tool" if hasattr(last, "tool_calls") and last.tool_calls else ENDgraph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tool", tool_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tool": "tool", END: END}) graph.add_edge("tool", "agent") app = graph.compile()运行: result = app.invoke({"messages": [HumanMessage(content="列出当前目录")]}) print(result["messages"][-1].content)Agent Memory 分级架构 高 Token 消耗的核心问题是全量注入记忆。推荐分三级: L1 工作记忆 → 最近 10~20 条消息,直接进 Prompt L2 任务记忆 → 当前项目/任务相关,按需读取 L3 长期记忆 → 用户偏好、历史经验,向量检索 TopKMemory Router 方案 不让 Memory 全量注入,而是先路由: def memory_router(state): query = state["query"] if "合约" in query: return {"memory_keys": ["web3"]} if "SpringBoot" in query: return {"memory_keys": ["coding", "project"]} return {"memory_keys": []}再根据 memory_keys 去向量库检索 Top-K 相关记忆注入 Prompt。 推荐存储方案 Redis → L1 短期记忆(最近消息、任务状态) pgvector → L2/L3 长期记忆(向量检索) Qdrant → 替代 pgvector(独立向量库)这样单次请求上下文通常能从 50k+ Token 压到 3k~10k Token,且效果往往更好(减少无关记忆干扰)。 OpenClaw 与 LangGraph 的关系 OpenClaw 是一个 Agent Runtime(Agent 操作系统),包含 Gateway(消息入口)、Skills(工具插件)、Memory、Workspace。LangGraph 提供了其中 Agent Loop + Persistence + Multi-Agent 的核心能力,但不负责 Gateway 和 Tool 生态。 用 LangGraph 可以实现 OpenClaw 绝大部分核心功能,真正耗工程量的是:权限系统、Tool 生态、Browser 自动化和可观测性系统。 最小技术栈(个人项目) Next.js / SpringBoot → 前后端 LangGraph → Agent 编排 Redis → 短期记忆 Postgres + pgvector → 长期记忆 Playwright → 浏览器操作 Quartz / @Scheduled → 定时任务
OpenCV 识别户型图中的墙线与柱子:HoughLines、轮廓检测与 YOLO
图片/CAD 图纸方案 墙线识别 import cv2 import numpy as npimg = cv2.imread("floorplan.png") gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)# 边缘检测 edges = cv2.Canny(gray, 50, 150, apertureSize=3)# 霍夫直线检测 lines = cv2.HoughLinesP( edges, rho=1, theta=np.pi / 180, threshold=100, minLineLength=50, maxLineGap=10 )if lines is not None: for line in lines: x1, y1, x2, y2 = line[0] cv2.line(img, (x1, y1), (x2, y2), (0, 255, 0), 2)minLineLength 控制最短检测长度,maxLineGap 控制允许的断线间距。 圆柱识别 circles = cv2.HoughCircles( gray, cv2.HOUGH_GRADIENT, dp=1, minDist=20, param1=50, param2=30, minRadius=10, maxRadius=100 )if circles is not None: circles = np.uint16(circles[0]) for cx, cy, r in circles: cv2.circle(img, (cx, cy), r, (255, 0, 0), 2)方柱/立柱识别 通过轮廓检测找矩形区域: contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)for cnt in contours: area = cv2.contourArea(cnt) if area < 200: continue # 多边形逼近 epsilon = 0.02 * cv2.arcLength(cnt, True) approx = cv2.approxPolyDP(cnt, epsilon, True) # 4 个顶点且面积合理 → 矩形 → 立柱 if len(approx) == 4: x, y, w, h = cv2.boundingRect(approx) aspect = w / h if h > 0 else 0 if 0.5 < aspect < 2.0: # 长宽比合理 cv2.rectangle(img, (x, y), (x + w, y + h), (0, 0, 255), 2)3D 点云方案(激光雷达/深度图) 墙面提取(RANSAC 平面分割) import open3d as o3dpcd = o3d.io.read_point_cloud("scan.ply")# RANSAC 平面分割 plane_model, inliers = pcd.segment_plane( distance_threshold=0.01, ransac_n=3, num_iterations=1000 )wall_cloud = pcd.select_by_index(inliers) other_cloud = pcd.select_by_index(inliers, invert=True)重复调用提取多个平面(多面墙体)。 圆柱提取(PCL RANSAC) PCL 提供 SACMODEL_CYLINDER 直接拟合圆柱: pcl::SACSegmentationFromNormals<PointT, NormalT> seg; seg.setModelType(pcl::SACMODEL_CYLINDER); seg.setMethodType(pcl::SAC_RANSAC); seg.setDistanceThreshold(0.05); seg.segment(*inliers, *coefficients);Python 可通过 python-pcl 或 Open3D 的 cylinder_fit 实现。 实时摄像头方案(推荐) 传统几何算法在以下情况容易失效:图纸断线、噪声 光照变化 非标准图纸格式推荐直接用 YOLO 检测/分割模型: from ultralytics import YOLOmodel = YOLO("yolov8n-seg.pt")# 训练自定义数据集(wall / pillar / cylinder 三类) model.train(data="floorplan.yaml", epochs=100, imgsz=640)# 推理 results = model("room.jpg") results[0].show()YOLOv8-seg 直接输出分割 Mask,比 HoughLines 对真实环境更鲁棒。 方案选型建议场景 推荐方案标准黑白 CAD 图 Canny + HoughLinesP平面图圆柱 HoughCircles复杂/真实户型图 轮廓检测 + 面积过滤激光雷达点云 RANSAC 平面/圆柱拟合实时摄像头 YOLOv8-seg 训练定制模型实际项目中通常是"深度学习检测 + 几何后处理"组合使用,单纯依赖传统算法稳定性较差。
PyInstaller 打包 FastAPI + Uvicorn:启动文件写法与依赖收集
直接打包 uvicorn main:app 的方式容易出现模块找不到的问题,推荐单独写入口文件。 项目结构 project/ ├── main.py # FastAPI app 定义 ├── start.py # PyInstaller 打包入口 ├── templates/ # Jinja2 模板(如有) └── static/ # 静态文件(如有)入口文件 start.py 关键点:直接传 app 对象,不用字符串 import uvicorn from main import appif __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)字符串形式 uvicorn.run("main:app", ...) 在 PyInstaller 环境中无法动态导入,会报: Error loading ASGI app. Could not import module "main"打包命令 基础打包: pyinstaller -F start.pyFastAPI/Uvicorn 有大量动态导入,需要显式收集: pyinstaller -F start.py \ --collect-all fastapi \ --collect-all uvicorn \ --collect-all starlette \ --collect-all pydantic包含模板和静态文件(macOS/Linux 用 :,Windows 用 ;): # Linux/macOS pyinstaller -F start.py \ --collect-all fastapi \ --collect-all uvicorn \ --collect-all starlette \ --collect-all pydantic \ --add-data "templates:templates" \ --add-data "static:static"# Windows pyinstaller -F start.py ^ --collect-all fastapi ^ --collect-all uvicorn ^ --collect-all starlette ^ --collect-all pydantic ^ --add-data "templates;templates" ^ --add-data "static;static"打包后静态文件路径 打包后程序解压到临时目录,__file__ 路径会变。读取打包内的文件需要: import sys import osdef get_resource_path(relative_path): if getattr(sys, 'frozen', False): base = sys._MEIPASS # PyInstaller 临时目录 else: base = os.path.dirname(__file__) return os.path.join(base, relative_path)# 挂载静态文件 from fastapi.staticfiles import StaticFiles app.mount("/static", StaticFiles(directory=get_resource_path("static")), name="static")常见错误 ModuleNotFoundError: No module named 'uvicorn.loops' --collect-all uvicornpydantic validator 找不到 --collect-all pydanticmultiprocessing 相关报错 确保 start.py 有 if __name__ == "__main__": 保护,避免多进程递归启动。 生产环境建议 PyInstaller 打包适合桌面工具分发,生产服务器更推荐:Docker 容器(环境固定、更新方便) systemd + venv(直接部署,无需打包) Supervisor 管理进程打包后可执行文件体积通常在 50-150 MB,启动时间比直接运行慢 3-10 秒。
PyInstaller 打包 FastAPI + uvicorn:能跑起来的最小写法
想把 FastAPI 项目打包成一个 exe 发给非技术用户,直接: pyinstaller -F main.py跑一下崩得七七八八。因为 uvicorn 走的是字符串式导入,PyInstaller 环境下拿不到;再加上一堆子模块(uvicorn.loops.*、starlette.*)静态分析捞不全。 正确起步:写个 start.py 不要直接打 main.py,写个显式导入 app 对象的启动文件: # start.py from main import app # 直接拿对象,不用字符串 import uvicornif __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)对应的 main.py: from fastapi import FastAPI app = FastAPI()@app.get("/") async def root(): return {"hello": "world"}打包 start.py: pyinstaller -F start.pyCould not import module "main" 的原因 如果 start.py 里写: uvicorn.run("main:app", host="0.0.0.0", port=8000)main:app 是字符串,uvicorn 会在运行时 importlib.import_module("main")。PyInstaller 打包后 main 不在 sys.path 里,直接崩: Error loading ASGI app. Could not import module "main"改成直接 import + 传对象最省事: from main import app uvicorn.run(app, ...)或者告诉 PyInstaller 也带上 main: pyinstaller -F start.py --hidden-import=mainuvicorn 子模块丢失 跑起来抛: ModuleNotFoundError: No module named 'uvicorn.loops'uvicorn 有一堆按需加载的 loop / protocol / logger 子模块,PyInstaller 全漏了。一次性带上: pyinstaller -F start.py \ --collect-all fastapi \ --collect-all uvicorn \ --collect-all starlette \ --collect-all pydantic--collect-all X = 把 X 及其所有子模块、数据文件、二进制全打进去。 静态文件 / 模板 用了 StaticFiles 或 Jinja2Templates 得加数据: Windows: pyinstaller -F start.py ^ --add-data "templates;templates" ^ --add-data "static;static"Linux/macOS: pyinstaller -F start.py \ --add-data "templates:templates" \ --add-data "static:static"代码里读路径要兼容 PyInstaller 临时目录: import sys, os from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templatesdef base_path(): return getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__)))app.mount("/static", StaticFiles(directory=os.path.join(base_path(), "static")), name="static") templates = Jinja2Templates(directory=os.path.join(base_path(), "templates"))后台运行、无 CMD 窗口 想双击不弹黑框: pyinstaller -F -w start.py --collect-all uvicorn ...-w 关控制台。注意:关了控制台,print / uvicorn 的日志就看不到了,改成写日志文件: uvicorn.run( app, host="0.0.0.0", port=8000, log_config={ "version": 1, "handlers": { "file": {"class": "logging.FileHandler", "filename": "app.log"}, }, "root": {"handlers": ["file"], "level": "INFO"}, }, )完整命令模板 一次搞定: pyinstaller -F -w start.py ^ --name myapi ^ --icon app.ico ^ --add-data "templates;templates" ^ --add-data "static;static" ^ --collect-all fastapi ^ --collect-all uvicorn ^ --collect-all starlette ^ --collect-all pydantic产物:dist/myapi.exe,双击就是一个后台服务。 什么时候别打包生产环境线上服务:Docker / systemd 更好 要热更新:exe 每次改代码都要重打 依赖 GPU / CUDA:打包体动辄几百 MB 多用户 / 高并发:exe 里塞 uvicorn 效率一般PyInstaller 更适合:内网小工具、给客户发一个"点两下就跑"的本地 API、演示 demo。 一句话总结 写 start.py + 直接 import app + --collect-all 四个关键字。字符串导入是坑,把 uvicorn 家族收全就通了。
Python str.translate 自定义 Base64:字符替换编解码原理
在逆向接口参数时,经常遇到这类代码: b64 = base64.b64encode(data.encode("utf-8")).decode("ascii") return b64.translate(TRANS)b64.translate(TRANS) 是 Python 字符串的内置 translate() 方法,不是某个库的 API。 translate() 做了什么 先 Base64 编码,再按照 TRANS 映射表替换字符。 translate() 一次可以替换多个字符,比链式 replace() 更高效: TRANS = str.maketrans({ "+": "-", "/": "_", })"a+b/c=".translate(TRANS) # → "a-b_c="TRANS 的常见形式 替换少量字符 TRANS = str.maketrans({ "+": "-", "/": "_", })这是标准的 URL Safe Base64 变体(RFC 4648),把 +/ 换成 -_,避免 URL 编码问题。 整体字符映射(自定义 Base64) src = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" dst = "NOPQRSTUVWXYZABCDEFGHIJKLMnopqrstuvwxyzabcdefghijklm0123456789-_"TRANS = str.maketrans(src, dst)这是在对整个 Base64 字符集做轮换映射,常用于接口参数混淆。 解码:把映射反过来 编码是 src → dst,解码就是 dst → src: REVERSE_TRANS = str.maketrans(dst, src)完整编解码实现 import base64src = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" dst = "NOPQRSTUVWXYZABCDEFGHIJKLMnopqrstuvwxyzabcdefghijklm0123456789-_"TRANS = str.maketrans(src, dst) REVERSE_TRANS = str.maketrans(dst, src)def encode(data: str) -> str: b64 = base64.b64encode(data.encode()).decode() return b64.translate(TRANS)def decode(data: str) -> str: raw = data.translate(REVERSE_TRANS) return base64.b64decode(raw).decode()x = encode("hello world") print(x) # 输出混淆后的字符串 print(decode(x)) # → hello world简单的 URL Safe Base64 变体 如果 TRANS 只替换 +/=,编解码更简单: import base64def url_safe_encode(data: str) -> str: b64 = base64.b64encode(data.encode()).decode() return b64.translate(str.maketrans({"+": "-", "/": "_", "=": ""}))def url_safe_decode(data: str) -> str: # 补齐 padding padded = data + "=" * (-len(data) % 4) raw = padded.translate(str.maketrans({"-": "+", "_": "/"})) return base64.b64decode(raw).decode()逆向时如何确认 TRANS 内容 遇到 b64.translate(TRANS) 时,找到 TRANS 的定义:如果是 str.maketrans(src, dst):把 dst 作为新的 src,src 作为新的 dst,即可反向还原 如果是 str.maketrans({...}):把字典的键值互换即可很多登录接口、签名参数都会用这种方式,目的是避免 +/= 在 URL 中被转义,或者简单混淆防止直接调用。
Python str.translate + Base64 = 常见的 URL-Safe 变体或混淆
看到 Python 代码里: b64 = base64.b64encode(data.encode("utf-8")).decode("ascii") return b64.translate(TRANS)这里的 .translate() 不是 Base64 库的方法——是 Python 字符串内置的 str.translate()。整个逻辑其实两步:b64encode 把数据编成标准 Base64 字符串 translate(TRANS) 按映射表把字符逐个替换常见用途:URL-Safe Base64、简单混淆、自定义字母表。 str.translate 是什么 str.translate(table) 按照 table 里的映射一次性替换多个字符——比一堆 .replace() 快得多: TRANS = str.maketrans({"a": "x", "b": "y"}) "abc".translate(TRANS) # "xyc"maketrans 三种参数形式: # 字典:字符 → 字符 str.maketrans({"a": "x", "b": "y"})# 两个字符串:一一对应 str.maketrans("abc", "xyz") # a→x, b→y, c→z# 三个参数:第三个是"要删除的字符" str.maketrans("", "", "aeiou") # 删除所有元音 "hello world".translate(...) # "hll wrld"底层是 dict[int, int|None],translate 每个字符按 code point 查表。 URL-Safe Base64(最常见用途) 标准 Base64 用了 + 和 /,在 URL 里出现要 percent-encode(%2B %2F)。URL-safe 变体换掉这两个字符: TRANS = str.maketrans({"+": "-", "/": "_"})b64 = base64.b64encode(data).decode("ascii") url_safe = b64.translate(TRANS)其实 Python 标准库直接提供了: base64.urlsafe_b64encode(data).decode("ascii")自己写 translate 版本效果一样,一般是兼容其它平台的历史代码。 反向:URL-Safe → 标准 TRANS_BACK = str.maketrans({"-": "+", "_": "/"})standard_b64 = url_safe.translate(TRANS_BACK)# 补 padding(长度必须是 4 的倍数) padded = standard_b64 + "=" * (-len(standard_b64) % 4)data = base64.b64decode(padded)URL-safe Base64 通常省掉了 = padding——解码时得手动补。 完全自定义字母表 某些爬虫 / 反调试代码会用非标准字母表做混淆: CUSTOM_ALPHABET = "NOPQRSTUVWXYZABCDEFGHIJKLM" + \ "nopqrstuvwxyzabcdefghijklm" + \ "0123456789-_"STANDARD_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + \ "abcdefghijklmnopqrstuvwxyz" + \ "0123456789+/"TRANS = str.maketrans(STANDARD_ALPHABET, CUSTOM_ALPHABET)编码时把标准 Base64 输出通过 TRANS 转成"看起来不一样但结构相同"的字符串。解码时反向映射再走标准 b64decode。 这是很低强度的混淆——攻击者拿到映射表就完全还原了。但作为"不让日志里的密码一眼可读"够用。 性能对比 translate 一次映射多个字符,比 .replace() 快很多: import timeits = "a" * 10000t1 = timeit.timeit( "s.replace('a', 'x').replace('b', 'y')", globals={"s": s}, number=10000, )TRANS = str.maketrans({"a": "x", "b": "y"}) t2 = timeit.timeit( "s.translate(TRANS)", globals={"s": s, "TRANS": TRANS}, number=10000, )print(t1 / t2) # translate 一般快 3-5 倍尤其映射表大的时候差距明显。 常见坑 1. maketrans 键值长度必须一样(字符串形式) str.maketrans("abc", "xy") # 报错:长度不匹配 str.maketrans("abc", "xyz") # OK字典形式没这个限制。 2. translate 对 bytes 用 bytes 表 b"abc".translate(bytes.maketrans(b"a", b"x"))字符串和字节的 translate 不通用。 3. 删除字符时用 None TRANS = {ord("a"): None} "abc".translate(TRANS) # "bc"或者用 str.maketrans("", "", "a") 生成的表。 一句话总结 看到 str.translate(TRANS) + Base64 的组合,大概率是做 URL-safe Base64 或简单字母表混淆。Python 标准库直接有 urlsafe_b64encode,自己用 translate 主要是历史代码。真正的性能优势在于批量替换比多次 replace 快得多。
Windows 报 WinError 206 路径太长?三种彻底解法
Python 里跑数据集处理,Windows 突然崩: FileNotFoundError: [WinError 206] 文件名或扩展名太长。: 'cache\\vendor\\dataset_2026\\...\\CAM_FRONT_WIDE_H110_HR\\virtual_image'不是数据太大,是这个路径字符串太长——超过了 Windows 老古董的 MAX_PATH = 260 限制。分批读取、分段处理、分次遍历都没用,open() 那一刻就已经炸了。 方案一:开注册表长路径支持(一次性) Windows 10/11 内核其实支持长路径,但默认关着。管理员 PowerShell: New-ItemProperty ` -Path "HKLM:\SYSTEM\CurrentControlSet\Control\FileSystem" ` -Name "LongPathsEnabled" ` -Value 1 ` -PropertyType DWORD ` -Force或 CMD: reg add HKLM\SYSTEM\CurrentControlSet\Control\FileSystem ^ /v LongPathsEnabled /t REG_DWORD /d 1 /f重启电脑,或者至少重启 Python / IDE 让新进程读到设置。 注意:仅这一步对某些老 Python API、部分 C 扩展、zipfile、shutil 不一定生效——它们内部还在走 260 限制的老 API。 方案二:\\?\ 前缀(最稳) Windows 内核 API 支持超长路径,但要显式加 \\?\ 前缀: import osdef win_long_path(path: str) -> str: """把路径转成支持超长的形式""" p = os.path.abspath(path) if not p.startswith("\\\\?\\"): p = "\\\\?\\" + p return pwith open(win_long_path(long_path)) as f: ...关键点:必须是绝对路径,\\?\ 不认相对路径 全程用反斜杠 \,不能混用正斜杠 UNC 路径(\\server\share\...)要写成 \\?\UNC\server\share\...这是数据管线、爬虫抓文件时最常用的做法,不依赖系统设置。 方案三:从源头缩短 工程上最优雅的还是让路径变短,两个方向: 把项目往盘根挪 D:\workspace\company_project\data\cache\... ❌ D:\d\cache\... ✅ 少几十字符缩短目录命名 自动生成的目录名往往长得离谱: CAM_PBQ_FRONT_WIDE_RESET_OPTICAL_H110_HR ❌ CAM1 ✅用一个 mapping 文件记录原名 → 短名即可。 npm / git 也会踩 不止 Python:git clone 遇到超深仓库 → 加: git config --system core.longpaths truenpm install / pnpm install 的 node_modules 层级太深 → 系统开 LongPathsEnabled + 项目挪到盘根一句话总结 **能开注册表就开、能加 \\?\ 就加、能挪盘根就挪。** 三管齐下才彻底摆脱 WinError 206。
YOLO 推理结果解读:置信度阈值、NMS 调参与工程部署建议
YOLO 推理输出结构 标准 YOLO API 输出格式(以 Ultralytics 为例): { "class": "vehicle", "class_id": 1, "confidence": 0.91, "bbox": { "x1": 156.15, "y1": 487.68, "x2": 653.12, "y2": 690.16, "width": 496.97, "height": 202.47 } }判断结果质量时关注两点:置信度分布:主要目标是否集中在 0.7 以上 框的空间分布:是否存在高度重叠的框四类常见问题 1. 重复框(NMS 不净) 同一辆车被检测 2~3 次,框坐标相近但不完全重合。原因是 IOU 阈值偏低,NMS 未能合并相似框。 # 默认 0.45 偏低,调高到 0.55~0.60 results = model(img, iou=0.55)2. 低置信度噪声 大量 confidence < 0.4 的框混入结果,尤其是行人、远处小目标。默认 conf=0.25 会保留大量噪声。 # 推荐生产阈值 results = model(img, conf=0.45)效果:去掉垃圾框,结果更干净,后续处理负担更小。 3. 小目标识别差 行人、远处目标置信度普遍偏低(< 0.5),框也不稳定。可能的原因:输入分辨率不够(模型内部将图缩放到 640×640 后,远处目标只占几个像素) 训练数据中远距离目标样本少优化方向: # 提高推理分辨率 results = model(img, imgsz=1280)或使用更适合小目标的模型(yolo11x 或 RT-DETR)。 4. 类别冲突(person vs two_wheel) 人骑车时同时触发 person 框和 two_wheel 框,两框高度重叠。后处理规则: def resolve_conflict(detections, iou_thresh=0.7): persons = [d for d in detections if d["class"] == "person"] bikes = [d for d in detections if d["class"] == "two_wheel"] to_remove = set() for p in persons: for b in bikes: if compute_iou(p["bbox"], b["bbox"]) > iou_thresh: # 保留置信度高的 if p["confidence"] < b["confidence"]: to_remove.add(id(p)) else: to_remove.add(id(b)) return [d for d in detections if id(d) not in to_remove]工程状态推荐参数参数 实验值 推荐生产值conf_threshold 0.25 0.45iou_threshold 0.45 0.55后处理 Pipeline results = model(img, conf=0.45, iou=0.55) detections = parse_results(results)# 1. 过滤极小框(去远处误检) detections = [d for d in detections if d["bbox"]["width"] * d["bbox"]["height"] > min_area]# 2. 类别冲突解决 detections = resolve_conflict(detections)# 3. 业务逻辑处理模型"能用"和"可上线"之间的差距主要就是以上这几个阈值和后处理步骤。
Python 自动安装 Chrome 插件:策略强制 vs 开发者模式加载
写脚本自动帮机器装个 Chrome 插件——听着简单。Chrome 团队故意不让你静默装本地 crx(安全策略,不是技术问题)。想绕开有几条现实可行的路。 路线一:企业策略强制安装(Web Store 上架的插件) Chrome 支持通过 Windows 注册表 / macOS plist / Linux JSON 策略,指定"必须自动装某扩展"。这是 IT 部门批量部署的标准做法。 前提是扩展在 Chrome Web Store 上架——策略只接受 update URL 拉取,不认本地 crx。 Windows 注册表方式 import winregEXT_ID = "aapocclcgogkmnckokdopfmhonfmgoek" # 举例:Google Docs Offline UPDATE_URL = "https://clients2.google.com/service/update2/crx"path = r"SOFTWARE\Policies\Google\Chrome\ExtensionInstallForcelist"key = winreg.CreateKey(winreg.HKEY_LOCAL_MACHINE, path) winreg.SetValueEx(key, "1", 0, winreg.REG_SZ, f"{EXT_ID};{UPDATE_URL}") winreg.CloseKey(key)写完重启 Chrome,扩展会自动安装、无法卸载。 注意:需要管理员权限 ExtensionInstallForcelist 的编号("1"、"2" ...)要不重复 只支持 Web Store 上架的扩展;企业内部扩展需要自建 update.xml配套删除: import winreg key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, path, 0, winreg.KEY_WRITE) winreg.DeleteValue(key, "1")路线二:Selenium 启动时加载本地插件 不做正式安装,只是启动 Chrome 时把插件目录加载进来——最适合自动化脚本、爬虫、批量测试。 加载解压后的目录(源码形式): from selenium import webdriver from selenium.webdriver.chrome.options import Optionsopts = Options() opts.add_argument(r"--load-extension=D:\extensions\my-plugin")driver = webdriver.Chrome(options=opts) driver.get("https://example.com")加载已打包的 crx: opts = Options() opts.add_extension(r"D:\extensions\my-plugin.crx")局限:不会真的"安装",关闭浏览器就没了 需要 Selenium 控制的 Chrome,不影响用户日常浏览器 Chrome 每次启动会有"以开发者模式使用未打包扩展"警告条路线三:Playwright 加载扩展 Playwright 加载扩展要用 persistent context,且必须用 Chromium(Chrome 分发版不能通过所有测试): from playwright.sync_api import sync_playwrightwith sync_playwright() as p: ctx = p.chromium.launch_persistent_context( user_data_dir="D:/tmp/pw-profile", headless=False, args=[ "--disable-extensions-except=D:\\extensions\\my-plugin", "--load-extension=D:\\extensions\\my-plugin", ], ) page = ctx.new_page() page.goto("https://example.com")不推荐的路线 UI 自动化拖 crx 到 chrome://extensions/:Chrome 会弹确认框、快捷键会变、chrome 版本升级后经常挂。别在这个方向上死磕。 改本地 Extensions 目录里的 Preferences JSON:Chrome 会检测校验和,第二次启动直接把你写进去的扩展禁用掉。 场景对应目的 用什么公司/网吧批量部署 Web Store 扩展 策略强制安装爬虫 / 自动化脚本用自己写的扩展 Selenium 加载需要长期挂着扩展,用户不能卸载 策略强制只是测试自己开发中的插件 Chrome 开发者模式手动一句话总结 要"真安装"就上企业策略(Web Store 扩展),要"临时加载"就 Selenium/Playwright --load-extension。Chrome 就是不让你静默装本地 crx,别硬撞。
Python 移动 / 复制目录的正确写法:shutil 全场景速查
Python 里"移动整个目录"或"把文件全复制到另一处"这类操作,全部靠 shutil。列几个最常用的模板。 移动整个目录 99% 场景直接: import shutilshutil.move(r"C:\source_dir", r"D:\target_dir")行为:目标不存在 → 直接改名/移动 目标已存在 → 源被移到 目标\源目录名 同盘 是秒移(本质是 rename) 跨盘 会退化成复制 + 删除想强制覆盖已有目标: import os, shutilsrc = r"C:\source_dir" dst = r"D:\target_dir\source_dir"if os.path.exists(dst): shutil.rmtree(dst) shutil.move(src, dst)复制整个目录(含子目录) Python 3.8+: import shutilshutil.copytree( r"C:\source_dir", r"D:\target_dir", dirs_exist_ok=True, # 目标已存在时合并写入 )会连带子目录、文件属性一起复制。 只复制当前目录里的文件(跳过子目录) import os, shutilsrc = r"C:\source_dir" dst = r"D:\target_dir" os.makedirs(dst, exist_ok=True)for name in os.listdir(src): s = os.path.join(src, name) if os.path.isfile(s): shutil.copy2(s, os.path.join(dst, name))copy2 会保留 mtime、权限,比 copy 更完整。 只复制某类文件 import os, shutilfor name in os.listdir(src): if name.endswith(".txt"): shutil.copy2( os.path.join(src, name), os.path.join(dst, name), )递归复制文件(扁平化,不保留目录结构) import os, shutilfor root, _, files in os.walk(src): for f in files: shutil.copy2(os.path.join(root, f), os.path.join(dst, f))注意同名文件会互相覆盖,要保结构就直接 copytree。 用 pathlib 版本(更现代) from pathlib import Path import shutilsrc = Path(r"D:\A\deep\folder") dst = Path(r"D:\B\deep\folder")if dst.exists(): shutil.rmtree(dst) dst.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(src, dst)Path 的 parent / mkdir(parents=True) 组合特别适合处理"目标父目录还没创建"的情况。 不要用 os.rename os.rename(src, dst)跨磁盘直接报错 不能覆盖已有目录除非明确是"同盘、无冲突",否则直接上 shutil.move。 一句话总结 移动用 shutil.move,整目录复制用 shutil.copytree(dirs_exist_ok=True)。其它场景在这两个基础上按需组合。
PyInstaller 打包 GUI 程序:无 CMD、文件夹形式
用 PyInstaller 打包一个 Python GUI 程序,想要"启动不弹 CMD 黑框、也不打成单个大文件",一条命令搞定: pyinstaller -w -D main.py参数解释 -w(或 --noconsole):不开控制台窗口。适合 Tkinter、PyQt、PySide、wxPython 之类的 GUI。 -D(或 --onedir):文件夹形式打包。产物结构: dist/ └── main/ ├── main.exe ├── python3x.dll └── ...(依赖 dll、pyd、资源)跟 -F 单文件相比,-D 启动快、崩溃少、易调试,就是需要发一整个文件夹。 加图标 pyinstaller -w -D -i app.ico main.py.ico 文件建议 256×256 或多尺寸 ico。 常见错误组合 GUI 忘了 -w: pyinstaller -D main.py # 会弹黑色 CMD 窗口GUI 只想单文件反而用了 -F -w: pyinstaller -F -w main.py能跑,但每次启动都要把打包体解压到临时目录,启动慢;sys._MEIPASS 拿资源要处理路径。 用 spec 文件更可控 第一次 pyinstaller 会生成 main.spec,之后修改这个文件重新构建: pyinstaller main.specspec 里控制"无 CMD"的字段: exe = EXE( pyz, a.scripts, name='main', console=False, # 无 CMD 窗口 icon='app.ico', ... )有 -w -D 起步足够,需要精细控制 hidden imports、数据文件、加签名时再上 spec。 加静态资源 程序里读了 templates/、assets/ 之类的目录,-D 只会打二进制,不带资源,运行时找不到。加 --add-data: Windows(分隔符 ;): pyinstaller -w -D main.py ^ --add-data "templates;templates" ^ --add-data "assets;assets"macOS / Linux(分隔符 :): pyinstaller -w -D main.py \ --add-data "templates:templates" \ --add-data "assets:assets"代码里读资源要用 PyInstaller 兼容路径: import sys, osdef resource_path(rel): base = getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__))) return os.path.join(base, rel)with open(resource_path("templates/index.html")) as f: ...找不到某个包 装了包但打包运行时抛 ModuleNotFoundError——PyInstaller 静态分析漏了动态导入。补 hidden-import: pyinstaller -w -D main.py --hidden-import=some_dynamic_module大量子模块可以直接 collect-all: pyinstaller -w -D main.py --collect-all fastapi --collect-all uvicorn一句话总结 pyinstaller -w -D main.py 是 GUI 程序打包最省心的一条命令。资源要 --add-data、动态导入要 --hidden-import 或 --collect-all。
Python 3.12+ 报 No module named 'distutils':三种恢复姿势
Python 3.13 环境下跑 labelImg,直接崩: File ".../site-packages/labelImg/labelImg.py", line 5, in <module> import distutils.spawn ModuleNotFoundError: No module named 'distutils'uv / pip 装依赖时也可能顺带看到: Because there are no versions of distutils and you require distutils, we can conclude that your requirements are unsatisfiable.为什么消失了 distutils 从 Python 3.10 就被标注 deprecated,3.12 起彻底删除。它的角色被 setuptools 和 packaging 接管,但很多老库(labelImg、部分 setup.py、老工具链)里还是硬 import distutils.spawn 或 distutils.util。 方案一:让 setuptools 顶上 从 setuptools 60 开始,装 setuptools 时会顺手把 distutils 兼容层装回来: pip install --upgrade setuptools装完再跑一次原命令,八成的老库都能过。注意要装到 触发报错的那个 Python 里,虚拟环境别搞混。 方案二:改一行源码(最彻底) distutils.spawn.find_executable 其实就等价于 shutil.which。找到出错的文件: .venv\Lib\site-packages\labelImg\labelImg.py把顶部的: import distutils.spawn替换成: import shutil然后把用到的地方: distutils.spawn.find_executable("xxx") # 改成 shutil.which("xxx")一处不会漏。这个改法在 CI 环境里最省事——不用装额外依赖。 方案三:降回 3.11(保稳) 赶时间又不想改源码: py -3.11 -m venv .venv311 .\.venv311\Scripts\activate pip install labelImg labelImg3.11 保留 distutils,官方支持到 2027 年 10 月,够用很久。 三种方案怎么选场景 推荐自己项目、能改源码 方案二(shutil.which)老工具/labelImg 类第三方包 方案一 + 方案三兜底生产环境、要稳 方案三(Python 3.11)一句话总结 distutils 3.12 起被删。八成情况装 setuptools 就能过;不行就直接改 shutil.which;最保险的是回 Python 3.11。
Python logging 同时输出到文件和控制台
Python 的 logging.basicConfig 有一个常见坑:一旦指定了 filename,默认只写文件,不往控制台打印。 问题复现 logging.basicConfig( filename="app.log", level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s" )print("print 能看到") logging.info("logger 看不到") # 只写文件,控制台不显示同时输出到文件和控制台 需要显式添加 StreamHandler: import logging import sys from datetime import datetimedef init_logger(log_file="app.log"): logger = logging.getLogger() logger.setLevel(logging.INFO) fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") # 文件输出 fh = logging.FileHandler(log_file, encoding="utf-8") fh.setFormatter(fmt) logger.addHandler(fh) # 控制台输出 ch = logging.StreamHandler(sys.stdout) ch.setFormatter(fmt) logger.addHandler(ch) return logger或者用 basicConfig + 手动追加: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", filename="app.log", encoding="utf-8" )# 追加控制台 Handler console = logging.StreamHandler() console.setLevel(logging.INFO) console.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logging.getLogger().addHandler(console)带结构化参数的 log 方法 class MyService: def __init__(self): self.logger = logging.getLogger(self.__class__.__name__) def log(self, *args, **kwargs): msg = " ".join(str(a) for a in args) if kwargs: extra = " ".join(f"{k}={v!r}" for k, v in kwargs.items()) msg = f"{msg} | {extra}" self.logger.info(msg)# 使用 svc = MyService() svc.log("任务完成", task_id=42, duration=1.5) # 输出: 任务完成 | task_id=42 duration=1.5PyInstaller --noconsole 下看不到日志 用 --noconsole 打包的 exe 绑定到 Windows subsystem,stdout / stderr 不存在,StreamHandler 无输出目标。 解决方案: 1. 只写文件,用 tail 等工具实时查看 # PowerShell(Windows 自带,等价于 tail -f) Get-Content app.log -Tail 50 -Wait2. 自定义 GUI 日志窗口(Tkinter) import tkinter as tk from tkinter.scrolledtext import ScrolledTextclass TextHandler(logging.Handler): def __init__(self, widget): super().__init__() self.widget = widget def emit(self, record): msg = self.format(record) self.widget.insert(tk.END, msg + "\n") self.widget.see(tk.END)root = tk.Tk() text = ScrolledText(root, width=80, height=20) text.pack()handler = TextHandler(text) handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")) logging.getLogger().addHandler(handler)logging.info("程序启动") root.mainloop()3. 调试 / 发布双版本 :: build.bat pyinstaller --onefile --console --name=控制器_debug scripts.py pyinstaller --onefile --noconsole --name=控制器 scripts.py开发用 debug 版,用户用 release 版。
