python實(shí)現(xiàn)Android與windows局域網(wǎng)文件夾同步

更新時(shí)間：2023年09月28日 09:13:04 作者：九狼

這篇文章主要給大家詳細(xì)介紹了python實(shí)現(xiàn)Android與windows局域網(wǎng)文件夾同步,文中有詳細(xì)的代碼示例和圖文介紹,具有一定的參考價(jià)值,需要的朋友可以參考下

Obsidian搭建個(gè)人筆記

最近在使用Obsidian搭建個(gè)人云筆記

盡管我使用COS圖床+gitee實(shí)現(xiàn)了云備份，但是在Android上使的Obsidian備份有點(diǎn)麻煩。還好我主要是在電腦端做筆記，手機(jī)只是作為閱讀工具。

所以，我寫(xiě)一個(gè)局域網(wǎng)文件夾同步工具，來(lái)解決這個(gè)問(wèn)題。

傳輸速度很快

局域網(wǎng)文件互傳

Windows和Android之間實(shí)現(xiàn)局域網(wǎng)內(nèi)文件互傳有以下幾種協(xié)議

HTTP 協(xié)議

優(yōu)點(diǎn):

實(shí)現(xiàn)簡(jiǎn)單,客戶端和服務(wù)器都有成熟的庫(kù)
安全性較好,支持HTTPS加密
可以傳輸不同類(lèi)型的數(shù)據(jù),包括文件、文本等

缺點(diǎn):

傳輸效率比Socket等協(xié)議低
需要自行處理大文件分片上傳和下載

Socket 協(xié)議

優(yōu)點(diǎn):

傳輸效率高,特別適合傳輸大文件
建立連接簡(jiǎn)單快速

缺點(diǎn):

需要處理粘包問(wèn)題,協(xié)議較為復(fù)雜
沒(méi)有加密,安全性差
需要處理網(wǎng)絡(luò)狀態(tài)變化等異常

SFTP 協(xié)議

優(yōu)點(diǎn):

安全性好,基于SSH通道傳輸
支持直接映射為本地磁盤(pán)訪問(wèn)

缺點(diǎn):

實(shí)現(xiàn)較復(fù)雜,需要找到可用的SFTP庫(kù)
傳輸效率比Socket低

WebSocket 協(xié)議

優(yōu)點(diǎn):

傳輸效率高,支持雙向通信
接口簡(jiǎn)單統(tǒng)一

缺點(diǎn):

需要處理連接狀態(tài),實(shí)現(xiàn)較為復(fù)雜
沒(méi)有加密,安全性較差

綜合來(lái)說(shuō),使用HTTP或Socket都是不錯(cuò)的選擇

WebSocket

但是最后我選擇了WebSocket，原因是Socket在處理接收數(shù)據(jù)的時(shí)候需要考慮緩沖區(qū)的大小和計(jì)算json結(jié)尾標(biāo)識(shí)，實(shí)現(xiàn)起來(lái)較為繁瑣，而WebSocket與Socket在實(shí)現(xiàn)這個(gè)簡(jiǎn)單的功能時(shí)的性能差別幾乎可以忽略不計(jì)，而且WebSocket可以輕松實(shí)現(xiàn)按行讀取數(shù)據(jù)，有效避免數(shù)據(jù)污染和丟失的問(wèn)題。最關(guān)鍵的一點(diǎn)是，WebSocket還可以輕松實(shí)現(xiàn)剪貼板同步功能。

我一開(kāi)始嘗試使用Socket來(lái)實(shí)現(xiàn)這個(gè)功能，但很快就發(fā)現(xiàn)實(shí)現(xiàn)起來(lái)相當(dāng)麻煩，于是換用了WebSocket，兩者在速度上沒(méi)有任何差別，用WebSocket起來(lái)舒服多了！

思路

使用Python將Windows目標(biāo)文件夾壓縮成zip格式，然后將其發(fā)送到Android設(shè)備。在Android設(shè)備上，接收壓縮文件后，通過(guò)MD5校驗(yàn)確保文件的完整性。一旦確認(rèn)無(wú)誤，將zip文件解壓到當(dāng)前目錄，最后刪除壓縮文件。整個(gè)過(guò)程既有趣又實(shí)用！

MD5校驗(yàn)沒(méi)寫(xiě)，一直用著也沒(méi)發(fā)現(xiàn)有壓縮包損壞的情況(超小聲)

定義json格式和功能標(biāo)識(shí)碼

為每個(gè)功能定義標(biāo)識(shí)碼

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ā)送完成");
}

用于文件傳輸過(guò)程中表示文件發(fā)送進(jìn)度的模型類(lèi)

data class FileSendingDot(
    val fileName: String,
    val bufferSize: Int,
    val total: Long,
    val sent: Long,
    val data: String
)

Python服務(wù)器端實(shí)現(xiàn)

創(chuàng)建websocket服務(wù)端

使用Python的asyncio和websockets模塊實(shí)現(xiàn)了一個(gè)異步的WebSocket服務(wù)器,通過(guò)異步事件循環(huán)來(lái)處理客戶端的連接和通信。

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()

解析同步請(qǐng)求,操作本地文件夾

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)分塊讀取文件并通過(guò)WebSocket發(fā)送每個(gè)數(shù)據(jù)塊,同時(shí)構(gòu)造消息對(duì)象封裝文件信息

file_data = f.read(buffer_size)
            sent_size += len(file_data)
            # 發(fā)送數(shù)據(jù)塊,包含序號(hào)和數(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 實(shí)現(xiàn)

請(qǐng)求所有文件訪問(wèn)權(quán)限

va launcher = registerForActivityResult(
ActivityResultContracts.StartActivityForResult()) { result ->
// 權(quán)限已授權(quán) or 權(quán)限被拒絕
}
private fun checkAndRequestAllFilePermissions() {
    //檢查權(quán)限
    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)建一個(gè)ActivityResultLauncher,用于啟動(dòng)并獲取文件夾選擇的回調(diào)結(jié)果。

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實(shí)現(xiàn)websocket

private val client = OkHttpClient.Builder().build()
//通過(guò)callbackFlow封裝,實(shí)現(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)
   //心跳機(jī)制
   launchHeartbeat()
    awaitClose { socket?.cancel() }
}.flowOn(Dispatchers.IO)
//服務(wù)端發(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到當(dāng)前文件夾

接收文件

顯示發(fā)送進(jìn)度

從FileSendingDot對(duì)象中取出已發(fā)送數(shù)據(jù)量sent和總數(shù)據(jù)量total。可以實(shí)時(shí)獲取文件傳輸?shù)倪M(jìn)度

用drawBehind在后面繪制矩形實(shí)現(xiàn)進(jìn)度條占位。根據(jù)進(jìn)度計(jì)算矩形寬度,實(shí)現(xiàn)進(jìn)度填充效果。不會(huì)遮擋子組件,很簡(jiǎn)潔地實(shí)現(xiàn)自定義進(jì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
    //通過(guò)判斷progress的值來(lái)決定是否顯示加載
    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ù)塊,包含序號(hào)和數(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ā)送結(jié)束標(biāo)志
    endmsg = {"type": "FILE_SENDEND", "msg": "發(fā)送完成", "data": "發(fā)送完成"}
    await ws.send(json.dumps(endmsg))
async def handle_client(websocket, path):
    # 用戶連接時(shí)打印日志
    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))
    # 用戶斷開(kāi)時(shí)打印日志
    print("用戶斷開(kāi)")
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實(shí)現(xiàn)Android與windows局域網(wǎng)文件夾同步的詳細(xì)內(nèi)容，更多關(guān)于python實(shí)現(xiàn)Android與windows文件同步的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: