一个解决 Apple 通用剪贴板失效的曲线救国方案
date
Oct 3, 2021
slug
yet-another-method-to-fix-universal-clipboard
status
Published
tags
fun
type
Post
outer_link
summary
利用 iOS 捷径 + VPS 实现文本消息 + 图片便捷发送到电脑端, 应该是最便捷的方案了.
1. 缘起
理论上讲, Apple 生态下的通用剪贴板 + AirDrop 应该是手机和电脑之间互相传送消息的最优解, 没有之一. 然而, 残酷的现实是, 通用剪贴板正常运作的概率在很多时候几乎是一个随机事件, 甚至有时候正常运作了居然还能带给我惊喜, 就离大谱.
与通用剪贴板不同, APNS (Apple 推送通知服务) 就异常稳定, 因此诞生了很多有用的工具, 如 Bark, Chanify 就是利用 APNS 将消息从电脑上推送到手机上, 并可以设置自动复制. 我尤其钟爱 Chanify, 除了对文字消息的支持, 还能发送图片和文件, 配合 Alfred Workflow 简直是方便到无敌.
那反过来呢? 如果想把消息快速从手机端发送到电脑端, 有没有这类快捷的方式呢?
其实手头如果有台 VPS, 配合 iOS 捷径是可以很方便地实现这一功能的:
使用 iOS 捷径在分享菜单里获取消息, 通过 http post 到 VPS 的 http server, VPS http server 同时也是一个 websocket client, 收到消息后进一步发送给 VPS 本机的 websocket server, 本机的 websocket server 再把消息广播给所有连接的 websocket client. 这里, 我们的本地电脑作为 websocket client, 事先已连接上 VPS, 在收到广播消息后执行消息复制等后处理就可以了.
2. 效果预览
我们先来看看效果.
(1) 场景 1: 文本、链接通过系统分享接口快捷发送到电脑端.
(2) 场景 2: 文本、链接通过小组件快捷发送到电脑端.
(3) 场景 3: 图像文件通过系统分享接口快捷发送到电脑端, 保存到 Downloads 文件夹下, 保存成功后并自动打开 finder.
3. Let's Do It!
条件准备
一台 VPS + 一点点 Python 代码 (以下采用 Python 3.9 环境配置, 其他版本也可以)
Python 依赖配置:
pip3 install -U flask websockets asyncio
Step1: Websocket server + client 配置
VPS 在 8183 端口开启一个 websocket 服务端, 收到任一客户端的消息后, 广播给其他所有客户端. 完整代码如下:
websocket_server.py (VPS 后台执行此脚本)
import os
import sys
import asyncio
import websockets
from pprint import pprint
import logging
logger = logging.getLogger(__name__)
logging.basicConfig(
stream=sys.stdout,
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
credentials = {
"username": "password",
}
clients = set()
class UserInfoProtocol(websockets.BasicAuthWebSocketServerProtocol):
async def check_credentials(self, username, password):
if username not in credentials:
print(f'invalid username: {username}')
return False
if credentials[username] != password:
print(f'invalid password {password} for user {username}')
return False
self.user = username
return True
def check_clients(websocket=None):
global clients
if websocket:
clients.add(websocket)
clients = set(x for x in clients if not x.closed)
async def broadcast_messages(msg):
global clients
check_clients()
websockets.broadcast(clients, msg)
async def main_logic(websocket, _):
check_clients(websocket)
user = websocket.user
print("User:", user)
print(f'Alive Clients: {len(clients)}')
msg = await websocket.recv()
await broadcast_messages(msg)
# the following requires Python >= 3.7
async def main():
async with websockets.serve(main_logic, "0.0.0.0", 8183, create_protocol=UserInfoProtocol):
await asyncio.Future() # run forever
if __name__ == "__main__":
asyncio.run(main())
注意:
代码中 L16 设置了用户名和密码, 默认用户名 username 和密码 password, 记得修改, 后同.
asyncio.run 是 Python ≥ 3.7 后的新语法, 低版本需要修改为:
start_server = websockets.serve(main_logic, "0.0.0.0", 8183, create_protocol=UserInfoProtocol)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
macOS (其实 win 本也可) 作为 websocket 客户端, 连接到 VPS websocket server. 逻辑为: 保持和服务器的连接, 实现断线重连 (默认 5s 缓冲期). 在收到 server 广播来的消息后 (json 消息), 调用 callback_fn 函数对消息做后处理.
完整代码如下:
websocket_client.py (电脑端后台执行此脚本)
import os
import sys
import asyncio
import websockets
import logging
import socket
import json
import base64
import time
logger = logging.getLogger(__name__)
logging.basicConfig(
stream=sys.stdout,
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
class WSClient():
# This is from https://gist.github.com/pgrandinetti/964747a9f2464e576b8c6725da12c1eb
def __init__(self, url, **kwargs):
self.url = url
# set some default values
self.reply_timeout = kwargs.get('reply_timeout') or 10
self.ping_timeout = kwargs.get('ping_timeout') or 5
self.sleep_time = kwargs.get('sleep_time') or 5
self.callback = kwargs.get('callback')
async def listen_forever(self):
while True:
# outer loop restarted every time the connection fails
logger.debug('Creating new connection...')
try:
async with websockets.connect(self.url) as ws:
while True:
# listener loop
try:
reply = await asyncio.wait_for(ws.recv(), timeout=self.reply_timeout)
except (asyncio.TimeoutError, websockets.exceptions.ConnectionClosed):
try:
pong = await ws.ping()
await asyncio.wait_for(pong, timeout=self.ping_timeout)
logger.debug('Ping OK, keeping connection alive...')
continue
except:
logger.debug(
'Ping error - retrying connection in {} sec (Ctrl-C to quit)'.format(self.sleep_time))
await asyncio.sleep(self.sleep_time)
break
logger.debug('Server said > {}'.format(reply))
if self.callback:
self.callback(reply)
except socket.gaierror:
logger.debug(
'Socket error - retrying connection in {} sec (Ctrl-C to quit)'.format(self.sleep_time))
await asyncio.sleep(self.sleep_time)
continue
except ConnectionRefusedError:
logger.debug('Nobody seems to listen to this endpoint. Please check the URL.')
logger.debug('Retrying connection in {} sec (Ctrl-C to quit)'.format(self.sleep_time))
await asyncio.sleep(self.sleep_time)
continue
def start_ws_client(client):
while True:
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(client.listen_forever())
except Exception as e:
# sometimes the network is down and the exception info is confusing. see https://github.com/aaugustin/websockets/issues/604
logger.debug('Exception: {}'.format(e))
logger.debug('Network is Done. Continue loop!')
time.sleep(5)
continue
def callback_fn(msg):
info = json.loads(msg)
if info["type"] == "txt":
data = info["data"]
print(f'rec txt from server: {data}')
cmd = f'echo "{data}" | pbcopy'
os.system(cmd)
elif info["type"] == "img":
img_b64 = info["b64"]
img_suffix = info["suffix"]
img_name = info["name"]
img_raw = base64.b64decode(img_b64.encode('utf-8'))
img_save_path = os.path.join(os.path.expanduser('~'), "Downloads", f"{img_name}.{img_suffix}")
print(f'rec img from server. saved to {img_save_path}')
with open(img_save_path, 'wb') as f:
f.write(img_raw)
jump_action = f'osascript -e "tell application \\"Finder\\"" -e activate -e "reveal POSIX file \\"{img_save_path}\\"" -e end tell'
os.system(jump_action)
if __name__ == "__main__":
ws_client = WSClient(url="ws://username:password@VPS_IP:8183", callback=callback_fn)
start_ws_client(ws_client)
L82: 对于文本消息, 将消息发送给 pbcopy 实现剪贴板自动复制 (pbcopy 是 macOS 剪贴板复制逻辑, 其他操作系统需要做相应修改).
L87: 对于图片消息, 获取图片 base64 编码数据, 图片名称和后缀, 将 base64 数据还原为图片后, 写入到 Downloads 文件夹下, 并调用 osascript 实现自动打开 Finder 的 Downloads 目录并显示在前台.
倒数第 2 行需要修改用户名、密码和 VPS 公网 IP.
Step2: HTTP Server 配置
VPS 在 8182 端口开启一个 http server, 并开放
/api/v1.0/sendtxt
和 /api/v1.0/sendimg
两个 restful api, 支持客户端通过 post 上传文本和图片. 同时, 该 http server 也是一个 websocket client, 收到 post 上传的消息后, 将消息序列化为 json 包发送给 VPS 本机的 websocket server, 从而触发本机 websocket server 的消息广播.完整代码如下:
flask_server.py (VPS 后台执行此脚本)
from flask import Flask, jsonify, request
import json
import websockets
import asyncio
import os
credentials = {
"username": "password",
}
clients = set()
app = Flask(__name__)
async def send_info(info):
uri = "ws://username:password@127.0.0.1:8183"
async with websockets.connect(uri) as websocket:
await websocket.send(info)
@app.route('/api/v1.0/sendtxt', methods=['POST'])
def task1():
user_name = request.form.get('username')
password = request.form.get('password')
if user_name in credentials and credentials[user_name] == password:
msg = request.form.get('msg')
info = json.dumps({"type": "txt", "data": msg})
asyncio.run(send_info(info))
return "OK"
@app.route('/api/v1.0/sendimg', methods=['POST'])
def task2():
user_name = request.form.get('username')
password = request.form.get('password')
if user_name in credentials and credentials[user_name] == password:
# iOS shorts will insert many '\n' in the base64 string
img_b64 = request.form.get('data').replace('\n', '')
img_suffix = request.form.get('type')
img_name = request.form.get('name')
info = json.dumps({"type": "img", "b64": img_b64, "suffix": img_suffix, "name": img_name})
asyncio.run(send_info(info))
return "OK"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8182)
L7 是 http server 的认证密码, L15 是作为 client 连接 websocket server 的认证密码, 这里作为演示设置二者保持一致.
Step3: iOS 捷径设置.
这里我一共设置了三个捷径:
其中
发送到 mac
对应场景 2 通过小组件手动输入文字. 分享文本到 mac
对应场景 1 通过分享菜单分享文字, 分享媒体到 mac
对应场景 3 通过分享菜单分享图片.这里以稍微复杂的
分享媒体到 mac
为例展开:点击展开截图:
通过共享表单获取图片输入, 然后将图像转换为 JPEG 减小体积并剔除敏感图片元信息, 最后将图像通过 base64 编码后 post 到 http server.
这里目前存在的局限性是: 由于对 iOS 捷径不够了解, 我并未找到更好的 post 图像文件的方式, 因此这里用 base64 编码后 post 到服务器. 当图片太大时, post 有一定几率失败 (可能是 iOS 捷径的限制, 具体原因不详, 待 debug).
4. 总结
个人认为此方式还是比较方便的.
优点:
- iOS 端不需要额外安装任何第三方客户端, 而且系统分享菜单使用体验极佳, 有时候(基本上)体验好于 AirDrop
- 操作简单, 编程上一共只需要 3 个 Python 脚本, VPS 运行一个 Websocket server 和一个 Flask Http Server, 电脑端只需要运行一个 Websocket client. 直接无脑 tmux 挂后台就好.
- 除了对文本的支持, 也支持图像等多媒体文件.
- 电脑端不限于 macOS, 只要能跑 Python 就行. 对于 Win + iPhone 搭配的用户非常方便.
- 总算不用打开微信文件传输助手了 (也算一个优点?)
缺点:
- 需要一台 VPS
其他:
- 以上脚本安全认证设置较为简单, 不过个人用问题不大
- 可以适当扩展, 如不采用现在的广播消息机制, 可以指定发送消息到哪台设备; 消息后处理也可以进一步扩展, 不只是局限于文本自动复制和图片自动保存.
- 本人对 iOS 捷径还是不够了解, 上传图像文件采用 base64 编码后 post, 可能还有更优方案.