python實現(xiàn)Android與windows局域網(wǎng)文件夾同步
更新時間：2023年09月28日 09:13:04 作者：九狼
這篇文章主要給大家詳細介紹了python實現(xiàn)Android與windows局域網(wǎng)文件夾同步,文中有詳細的代碼示例和圖文介紹,具有一定的參考價值,需要的朋友可以參考下
Obsidian搭建個人筆記
最近在使用Obsidian搭建個人云筆記
盡管我使用COS圖床+gitee實現(xiàn)了云備份，但是在Android上使的Obsidian備份有點麻煩。還好我主要是在電腦端做筆記，手機只是作為閱讀工具。
所以，我寫一個局域網(wǎng)文件夾同步工具，來解決這個問題。
傳輸速度很快
局域網(wǎng)文件互傳
Windows和Android之間實現(xiàn)局域網(wǎng)內(nèi)文件互傳有以下幾種協(xié)議
HTTP 協(xié)議
優(yōu)點:
- 實現(xiàn)簡單,客戶端和服務器都有成熟的庫
- 安全性較好,支持HTTPS加密
- 可以傳輸不同類型的數(shù)據(jù),包括文件、文本等
缺點:
- 傳輸效率比Socket等協(xié)議低
- 需要自行處理大文件分片上傳和下載
Socket 協(xié)議
優(yōu)點:
- 傳輸效率高,特別適合傳輸大文件
- 建立連接簡單快速
缺點:
- 需要處理粘包問題,協(xié)議較為復雜
- 沒有加密,安全性差
- 需要處理網(wǎng)絡狀態(tài)變化等異常
SFTP 協(xié)議
優(yōu)點:
- 安全性好,基于SSH通道傳輸
- 支持直接映射為本地磁盤訪問
缺點:
- 實現(xiàn)較復雜,需要找到可用的SFTP庫
- 傳輸效率比Socket低
WebSocket 協(xié)議
優(yōu)點:
- 傳輸效率高,支持雙向通信
- 接口簡單統(tǒng)一
缺點:
- 需要處理連接狀態(tài),實現(xiàn)較為復雜
- 沒有加密,安全性較差
綜合來說,使用HTTP或Socket都是不錯的選擇
WebSocket
但是最后我選擇了WebSocket，原因是Socket在處理接收數(shù)據(jù)的時候需要考慮緩沖區(qū)的大小和計算json結尾標識，實現(xiàn)起來較為繁瑣，而WebSocket與Socket在實現(xiàn)這個簡單的功能時的性能差別幾乎可以忽略不計，而且WebSocket可以輕松實現(xiàn)按行讀取數(shù)據(jù)，有效避免數(shù)據(jù)污染和丟失的問題。最關鍵的一點是，WebSocket還可以輕松實現(xiàn)剪貼板同步功能。
我一開始嘗試使用Socket來實現(xiàn)這個功能，但很快就發(fā)現(xiàn)實現(xiàn)起來相當麻煩，于是換用了WebSocket，兩者在速度上沒有任何差別，用WebSocket起來舒服多了！
思路
使用Python將Windows目標文件夾壓縮成zip格式，然后將其發(fā)送到Android設備。在Android設備上，接收壓縮文件后，通過MD5校驗確保文件的完整性。一旦確認無誤，將zip文件解壓到當前目錄，最后刪除壓縮文件。整個過程既有趣又實用！
MD5校驗沒寫，一直用著也沒發(fā)現(xiàn)有壓縮包損壞的情況(超小聲)
定義json格式和功能標識碼
為每個功能定義標識碼
enum class SocketType(val type: String, val msg: String) {
FILE_SYNC("FILE_SYNC", "文件同步"),
FOLDER_SYNC("FOLDER_SYNC", "文件夾同步"),
CLIPBOARD_SYNC("CLIPBOARD_SYNC", "剪貼板同步"),
HEARTBEAT("HEARTBEAT", "心跳"),
FILE_SENDING("FILE_SENDING", "發(fā)送中"),
FOLDER_SYNCING("FOLDER_SYNCING", "文件夾同步中"),
FILE_SENDEND("FILE_SENDEND", "發(fā)送完成");
}用于文件傳輸過程中表示文件發(fā)送進度的模型類
data class FileSendingDot(
val fileName: String,
val bufferSize: Int,
val total: Long,
val sent: Long,
val data: String
)Python服務器端實現(xiàn)
創(chuàng)建websocket服務端
使用Python的asyncio和websockets模塊實現(xiàn)了一個異步的WebSocket服務器,通過異步事件循環(huán)來處理客戶端的連接和通信。
import asyncio import websockets start_server = websockets.serve(handle_client, "", 9999) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever()
解析同步請求,操作本地文件夾
json_obj = json.loads(data)
type_value = json_obj["type"]
data_value = json_obj["data"]
if type_value == "FILE_SYNC":
await send_file(websocket,"FILE_SENDING", file_path)利用循環(huán)分塊讀取文件并通過WebSocket發(fā)送每個數(shù)據(jù)塊,同時構造消息對象封裝文件信息
file_data = f.read(buffer_size)
sent_size += len(file_data)
# 發(fā)送數(shù)據(jù)塊,包含序號和數(shù)據(jù)
send_file_data = base64.b64encode(file_data).decode()
file_seading_data = {
"fileName": filename,
"bufferSize":buffer_size,
"total": total_size,
"sent": sent_size,
"data": send_file_data,
}
msg = {
"type": type,
"msg": "發(fā)送中",
"data": json.dumps(file_seading_data),
}
await ws.send(json.dumps(msg))安卓客戶端 Jetpack ComposeUI 實現(xiàn)
請求所有文件訪問權限
va launcher = registerForActivityResult(
ActivityResultContracts.StartActivityForResult()) { result ->
// 權限已授權 or 權限被拒絕
}
private fun checkAndRequestAllFilePermissions() {
//檢查權限
if (!Environment.isExternalStorageManager()) {
val intent = Intent(Settings.ACTION_MANAGE_APP_ALL_FILES_ACCESS_PERMISSION)
intent.setData(Uri.parse("package:$packageName"))
launcher.launch(intent)
}
}自定義保存路徑
選擇文件夾
rememberLauncherForActivityResult() 創(chuàng)建一個ActivityResultLauncher,用于啟動并獲取文件夾選擇的回調(diào)結果。
val selectFolderResult = rememberLauncherForActivityResult(ActivityResultContracts.StartActivityForResult()) { data ->
val uri = data.data?.data
if (uri != null) {
intentChannel.trySend(ViewIntent.SelectFolder(uri))
} else {
ToastModel("選擇困難! ?(???)?", ToastModel.Type.Info).showToast()
}
}Uri的path
fun Uri.toFilePath(): String {
val uriPath = this.path ?: return ""
val path = uriPath.split(":")[1]
return Environment.getExternalStorageDirectory().path + "/" + path
}
okhttp實現(xiàn)websocket
private val client = OkHttpClient.Builder().build()
//通過callbackFlow封裝,實現(xiàn)流式API
fun connect() =
createSocketFlow()
.onEach {
LogX.i("WebSocket", "收到消息 $it")
}.retry(reconnectInterval)
private fun createSocketFlow(): Flow<String> = callbackFlow {
val request = Request.Builder()
.url("ws://192.168.0.102:9999")
.build()
val listener = object : WebSocketListener() {
...接收消息的回調(diào)
}
socket = client.newWebSocket(request, listener)
//心跳機制
launchHeartbeat()
awaitClose { socket?.cancel() }
}.flowOn(Dispatchers.IO)
//服務端發(fā)送數(shù)據(jù)
fun send(message: String) {
socket?.send(message)
}接收文件
使用 Base64.decode() 方法將 base64 數(shù)據(jù)解碼成字節(jié)數(shù)組 fileData
val fileName = dot.fileName val file = File(AppSystemSetManage.fileSavePath, fileName) val fileData = Base64.decode(dot.data, Base64.DEFAULT)
- 接著就是使用IO數(shù)據(jù)流
OutputStream加上自定義的路徑一頓操作 就得到zip文件了 - 最后解壓zip到當前文件夾
接收文件
顯示發(fā)送進度
從FileSendingDot對象中取出已發(fā)送數(shù)據(jù)量sent和總數(shù)據(jù)量total。 可以實時獲取文件傳輸?shù)倪M度
用drawBehind在后面繪制矩形實現(xiàn)進度條占位。根據(jù)進度計算矩形寬度,實現(xiàn)進度填充效果。不會遮擋子組件,很簡潔地實現(xiàn)自定義進度條。
Box(
modifier = Modifier
.fillMaxWidth()
.drawBehind {
val fraction = progress * size.width
drawRoundRect(
color = progressColor,
size = Size(width = fraction, height = size.height),
cornerRadius = CornerRadius(12.dp.toPx()),
alpha = 0.9f,
)
}@Composable
fun ProgressCard(
modifier: Modifier = Modifier,
title: String,
progress: Float,
onClick: () -> Unit = {}
) {
val progressColor = WordsFairyTheme.colors.themeAccent
//通過判斷progress的值來決定是否顯示加載
val load = progress > 0F
val textColor = if (load) WordsFairyTheme.colors.themeUi else WordsFairyTheme.colors.textPrimary
OutlinedCard(
modifier = modifier,
onClick = onClick,
colors =
CardDefaults.cardColors(WordsFairyTheme.colors.itemBackground),
border = BorderStroke(1.dp, textColor)
) {
Box(
modifier = Modifier
.fillMaxWidth()
.drawBehind {
val fraction = progress * size.width
drawRoundRect(
color = progressColor,
size = Size(width = fraction, height = size.height),
cornerRadius = CornerRadius(12.dp.toPx()),
alpha = 0.9f,
)
},
content = {
Row {
Title(
title = title, Modifier.padding(16.dp),
color = textColor
)
Spacer(Modifier.weight(1f))
if (load)
Title(
title = "${(progress * 100).toInt()}%", Modifier.padding(16.dp),
color = textColor
)
}
}
)
}
}效果圖
python代碼
import asyncio
import websockets
import os
from pathlib import Path
import pyperclip
import json
import base64
import zipfile
import math
FILE_BUFFER_MIN = 1024
FILE_BUFFER_MAX = 1024 * 1024 # 1MB
file_path = "E:\\xy\\FruitSugarContentDetection.zip"
folder_path = "E:\\Note\\Obsidian"
zip_path = "E:\\Note\\Obsidian.zip"
async def send_file(ws,type, filepath):
# 獲取文件名
filename = os.path.basename(filepath)
total_size = os.path.getsize(filepath)
sent_size = 0
if total_size < FILE_BUFFER_MAX * 10:
buffer_size = math.ceil(total_size / 100)
else:
buffer_size = FILE_BUFFER_MAX
with open(filepath, "rb") as f:
while sent_size < total_size:
file_data = f.read(buffer_size)
sent_size += len(file_data)
# 發(fā)送數(shù)據(jù)塊,包含序號和數(shù)據(jù)
send_file_data = base64.b64encode(file_data).decode()
file_seading_data = {
"fileName": filename,
"bufferSize":buffer_size,
"total": total_size,
"sent": sent_size,
"data": send_file_data,
}
msg = {
"type": type,
"msg": "發(fā)送中",
"data": json.dumps(file_seading_data),
}
await ws.send(json.dumps(msg))
print((sent_size / total_size) * 100)
# 發(fā)送結束標志
endmsg = {"type": "FILE_SENDEND", "msg": "發(fā)送完成", "data": "發(fā)送完成"}
await ws.send(json.dumps(endmsg))
async def handle_client(websocket, path):
# 用戶連接時打印日志
print("用戶連接")
async for data in websocket:
print(data)
json_obj = json.loads(data)
type_value = json_obj["type"]
data_value = json_obj["data"]
if type_value == "FILE_SYNC":
await send_file(websocket,"FILE_SENDING", file_path)
if type_value == "FOLDER_SYNC":
zip_folder(folder_path, zip_path)
await send_file(websocket,"FOLDER_SYNCING", zip_path)
if type_value == "CLIPBOARD_SYNC":
pyperclip.copy(data_value)
print(data_value)
if type_value == "HEARTBEAT":
dictionary_data = {
"type": "HEARTBEAT",
"msg": "hi",
"data": "",
}
await websocket.send(json.dumps(dictionary_data))
# 用戶斷開時打印日志
print("用戶斷開")
def zip_folder(folder_path, zip_path):
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
for root, _, files in os.walk(folder_path):
for file in files:
file_path = os.path.join(root, file)
zipf.write(file_path, arcname=os.path.relpath(file_path, folder_path))
start_server = websockets.serve(handle_client, "", 9999)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()github:https://github.com/JIULANG9/FileSync
gitee:https://gitee.com/JIULANG9/FileSync
以上就是python實現(xiàn)Android與windows局域網(wǎng)文件夾同步的詳細內(nèi)容，更多關于python實現(xiàn)Android與windows文件同步的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了python通過http下載文件的方法詳解,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-07-07
prompt_toolkit 是一個用于構建強大交互式命令行的 Python 工具庫。接下來通過本文給大家介紹Python?命令行?prompt_toolkit?庫的相關知識，感興趣的朋友一起看看吧2022-01-01
這篇文章主要介紹了pandas 如何分割字符的實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-07-07
這篇文章主要介紹了Django權限控制的使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-01-01
這篇文章主要介紹了Python抽象類的新寫法,本文講解了老版本中的hack方式實現(xiàn)抽象類,以及2.7以后使用abstractmethod模塊寫抽象類的方法,需要的朋友可以參考下2015-06-06
pyqt5實現(xiàn)繪制ui,列表窗口,滾動窗口顯示圖片的方法
今天小編就為大家分享一篇pyqt5實現(xiàn)繪制ui,列表窗口,滾動窗口顯示圖片的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-06-06
在數(shù)據(jù)處理和繪圖中,我們通常會遇到直線或曲線的擬合問題,本文主要介紹了Python實現(xiàn)線性擬合及繪圖的示例代碼,具有一定的參考價值,感興趣的可以了解一下2024-04-04
這篇文章主要介紹了Python?爬取微博熱搜頁面，關于Python?爬蟲，爬取網(wǎng)頁等相關內(nèi)容一般可作為小練習，下面文章Python?爬取微博熱搜頁面也如此，需要的小伙伴可以參考一下2022-01-01
Python Django項目和應用的創(chuàng)建詳解
這篇文章主要為大家介紹了Python Django項目和應用的創(chuàng)建，具有一定的參考價值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助2021-11-11
這篇文章主要介紹了python如何實時獲取tcpdump輸出，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2020-09-09