基於七牛Python SDK寫的乙個同步指令碼

2022-02-25 17:43:27 字數 4266 閱讀 2801

最近剛搭了個markdown靜態部落格,想把部落格的放到雲儲存中。

經過調研覺得七牛可以滿足我個人的需求,就選它了。

部落格要引用就要先將上傳到雲上。

雖然七牛**後台可以上傳檔案,但每次上傳都需要先登入,然後選擇,設定連線位址,才能上傳。

這個過程有些繁瑣,所以我便想用七牛雲提供的sdk寫個一同步工具,方便增量同步檔案。

有了這個想法,就馬上行動。花了大概乙個上午的時間,總算把這個工具給寫出來,

並放到gitosc和github上。

(的markdown**區顯示不友好,可以到我的個人部落格中瀏覽)

#!/usr/bin/env python

#-*- coding:utf-8 -*-

# # author = "heqingpan"

# author_email = "[email protected]"

# url = ""

import qiniu

from qiniu import auth

from qiniu import bucketmanager

import os

import re

access_key = ''

secret_key = ''

bucket_name = ''

bucket_domain = ''

q = auth(access_key, secret_key)

bucket = bucketmanager(q)

basedir=os.path.realpath(os.path.dirname(__file__))

filename=__file__

ignore_paths=[filename,"c".format(filename)]

ignore_names=[".ds_store",".git",".gitignore"]

charset="utf8"

diff_time=2*60

def list_all(bucket_name, bucket=none, prefix="", limit=100):

rlist=

if bucket is none:

bucket = bucketmanager(q)

marker = none

eof = false

while eof is false:

ret, eof, info = bucket.list(bucket_name, prefix=prefix, marker=marker, limit=limit)

marker = ret.get('marker', none)

for item in ret['items']:

if eof is not true:

# 錯誤處理

#print "error"

pass

return rlist

def get_files(basedir="",fix="",rlist=none,ignore_paths=,ignore_names=):

if rlist is none:

rlist=

for subfile in os.listdir(basedir):

temp_path=os.path.join(basedir,subfile)

tp=os.path.join(fix,subfile)

if tp in ignore_names:

continue

if tp in ignore_paths:

continue

if os.path.isfile(temp_path):

elif os.path.isdir(temp_path):

get_files(temp_path,tp,rlist,ignore_paths,ignore_names)

return rlist

def get_valid_key_files(subdir=""):

basedir=subdir or basedir

files = get_files(basedir=basedir,ignore_paths=ignore_paths,ignore_names=ignore_names)

return map(lambda f:(f.replace("\\","/"),f),files)

def sync():

qn_keys=list_all(bucket_name,bucket)

qn_set=set(qn_keys)

l_key_files=get_valid_key_files(basedir)

k2f={}

update_keys=

u_count=500

u_index=0

for k,f in l_key_files:

k2f[k]=f

str_k=k

if isinstance(k,str):

k=k.decode(charset)

if k in qn_set:

u_index+=1

if u_index > u_count:

u_index-=u_count

update_file(k2f,update_keys)

update_keys=

else:

# upload

upload_file(k,os.path.join(basedir,f))

if update_keys:

update_file(k2f,update_keys)

print "sync end"

def update_file(k2f,ulist):

ops=qiniu.build_batch_stat(bucket_name,ulist)

rets,infos = bucket.batch(ops)

for i in xrange(len(ulist)):

k=ulist[i]

f=k2f.get(k)

ret=rets[i]["data"]

size=ret.get("fsize",none)

put_time = int(ret.get("puttime")/10000000)

local_size=os.path.getsize(f)

local_time=int(os.path.getatime(f))

if local_size==size:

continue

if put_time >= local_time - diff_time:

# is new

continue

# update

upload_file(k,os.path.join(basedir,f))

def upload_file(key,localfile):

print "upload_file:"

print key

token = q.upload_token(bucket_name, key)

mime_type = get_mime_type(localfile)

params =

progress_handler = lambda progress, total: progress

ret, info = qiniu.put_file(token, key, localfile, params, mime_type, progress_handler=progress_handler)

def get_mime_type(path):

mime_type = "text/plain"

return mime_type

def main():

sync()

if __name__=="__main__":

main()

這個同步指令碼支援批量比較檔案,差異增量更新、批量更新。

access_key = ''

secret_key = ''

bucket_name = ''

註冊後可以拿到對應的資訊

python qiniusync.py
寫完提交之後才發現,七牛已經提供相應的工具,我這個算是重複造輪子吧。

既然已經寫,就發出來,當做熟悉一下七牛的sdk也不錯,說不定以後還能用的上。

的markdown**區顯示不友好,可以到我的個人部落格中瀏覽

基於七牛Python SDK寫的乙個同步指令碼

最近剛搭了個markdown靜態部落格,想把部落格的放到雲儲存中。經過調研覺得七牛可以滿足我個人的需求,就選它了。部落格要引用就要先將上傳到雲上。雖然七牛 後台可以上傳檔案,但每次上傳都需要先登入,然後選擇,設定連線位址,才能上傳。這個過程有些繁瑣,所以我便想用七牛雲提供的sdk寫個一同步工具,方便...

關於七層架構的重新理解

在進行重構的時候,對於七層懵懵懂懂,並不知曉分層的意義。隨著實踐的推移,有了新的理解,與大家分享。按照實際需求來講,我們 的架構分為ubd三層就夠了,其他的層是為了輔助這三層而產生的。u層,即介面層。用於直接展示給使用者的一層,是軟體與使用者交流最直接的一層。u層的作用是接受使用者給出的資料,並將相...

純Delphi 原生寫的 上傳到七牛的功能

分兩個函式 uploadtoqiniu和directuploadtoqiniu uploadtoqiniu這個函式使用分片,分段的方式上傳,並有上傳進度 採用多執行緒同時進行,該方法適用於上傳較大檔案。directuploadtoqiniu該函式直接使用form表單的形式上傳,沒有上傳進度 適用於上...