使用 Seafile,人们可以创建一个公共上传链接(例如
https://cloud.seafile.com/u/d/98233edf89/
),通过浏览器上传文件,无需身份验证。
Seafile webapi 不支持任何不带身份验证令牌的上传。
我如何从带有curl的命令行或从python脚本使用这种链接?
用curl花了2个小时才找到解决方案,需要两个步骤:
repo-id
作为查询参数向公共上行链路 url 发出 get 请求,如下所示:curl 'https://cloud.seafile.com/ajax/u/d/98233edf89/upload/?r=f3e30b25-aad7-4e92-b6fd-4665760dd6f5' -H 'Accept: application/json' -H 'X-Requested-With: XMLHttpRequest'
答案是 (json) 在下一个上传帖子中使用的 id 链接,例如:
{"url": "https://cloud.seafile.com/seafhttp/upload-aj/c2b6d367-22e4-4819-a5fb-6a8f9d783680"}
curl 'https://cloud.seafile.com/seafhttp/upload-aj/c2b6d367-22e4-4819-a5fb-6a8f9d783680' -F file=@./tmp/index.html -F filename=index.html -F parent_dir="/my-repo-dir/"
答案又是json,例如
[{"name": "index.html", "id": "0a0742facf24226a2901d258a1c95e369210bcf3", "size": 10521}]
完成;)
#!/usr/bin/env bash
# this script depend on jq,check it first
RED='\033[0;31m'
NC='\033[0m' # No Color
if ! command -v jq &> /dev/null
then
echo -e "${RED}jq could not be found${NC}, installed and restart plz!\n"
exit
fi
usage () { echo "Usage : $0 -u <username> -p <password> -h <seafile server host> -f <upload file path> -d <parent dir default value is /> -r <repo id> -t <print debug info switch off/on,default off>"; }
# parse args
while getopts "u:p:h:f:d:r:t:" opts; do
case ${opts} in
u) USER=${OPTARG} ;;
p) PASSWORD=${OPTARG} ;;
h) HOST=${OPTARG} ;;
f) FILE=${OPTARG} ;;
d) PARENT_DIR=${OPTARG} ;;
r) REPO=${OPTARG} ;;
t) DEBUG=${OPTARG} ;;
*) usage; exit;;
esac
done
# those args must be not null
if [ ! "$USER" ] || [ ! "$PASSWORD" ] || [ ! "$HOST" ] || [ ! "$FILE" ] || [ ! "$REPO" ]
then
usage
exit 1
fi
# optional args,set default value
[ -z "$DEBUG" ] && DEBUG=off
[ -z "$PARENT_DIR" ] && PARENT_DIR=/
# print vars key and value when DEBUG eq on
[[ "on" == "$DEBUG" ]] && echo -e "USER:${USER} PASSWORD:${PASSWORD} HOST:${HOST} FILE:${FILE} PARENT_DIR:${PARENT_DIR} REPO:${REPO} DEBUG:${DEBUG}"
# login and get token
TOKEN=$(curl -s --location --request POST "${HOST}/api2/auth-token/" --header 'Content-Type: application/x-www-form-urlencoded' --data-urlencode "username=${USER}" --data-urlencode "password=${PASSWORD}" | jq -r ".token")
[ -z "$TOKEN" ] && echo -e "${RED}login seafile faild${NC}, call your administrator plz!\n" && exit 1
# gen upload link
UPLOAD_LINK=$(curl -s --header "Authorization: Token ${TOKEN}" "${HOST}/api2/repos/${REPO}/upload-link/?p=${PARENT_DIR}" | jq -r ".")
[ -z "$UPLOAD_LINK" ] && echo -e "${RED}get upload link faild${NC}, call your administrator plz!\n" && exit 1
# upload file
UPLOAD_RESULT=$(curl -s --header "Authorization: Token ${TOKEN}" -F file="@${FILE}" -F filename=$(basename ${FILE}) -F parent_dir="${PARENT_DIR}" -F replace=1 "${UPLOAD_LINK}?ret-json=1")
[ -z "$UPLOAD_RESULT" ] && echo -e "${RED}faild to upload ${FILE}${NC}, call your administrator plz!\n" && exit 1
# print upload result
[[ "on" == "$DEBUG" ]] && echo -e "TOKEN:${TOKEN} UPLOAD_LINK:${UPLOAD_LINK} UPLOAD_RESULT:${UPLOAD_RESULT}"
另存为
seafile-upload.sh
# ubuntu
apt install -y jq
# centos
# yum install -y jq
chmod +x ./seafile-upload.sh
./seafile-upload.sh -u <username> -p <password> -h <seafile server host> -f <upload file path> -d <parent dir default value is /,must be start with /> -r <repo id> -t <print debug info switch off/on,default off>
根据我对浏览器中文件上传行为的观察,我编写了一个用于命令行使用的 Python 脚本。只需要上传页面链接(由 Seafile 生成)和本地文件路径。
我还将脚本作为 GitHub gist 上传。
设置:
requests
和 beautifulsoup4
。requests_toolbelt
和tqdm
以查看实时上传进度。usage: upload_seafile.py [-h] -l LINK -f FILE [FILE ...] [--verbose]
options:
-h, --help show this help message and exit
-l LINK, --link LINK upload page link (generated by seafile)
-f FILE [FILE ...], --file FILE [FILE ...]
file(s) to upload
--verbose show detailed output
关键逻辑:
parent_dir
和token
(无需登录)。parent_dir
参数。代码:
import re
import argparse
import sys
import copy
from pathlib import Path
from urllib.parse import urlparse
import requests
from bs4 import BeautifulSoup
optional_packages = True
try:
# optional, for upload progess updates
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from tqdm import tqdm
except ImportError:
optional_packages = False
def extract_var(script_text, variable_name, default=None):
if variable_name in script_text:
# match: var_name: "value" or var_name: 'value' or var_name = "value" or var_name = 'value'
pattern = re.compile(r'{}\s*[:=]\s*(["\'])(.*?)\1'.format(re.escape(variable_name)))
match = pattern.search(script_text)
if match:
return match.group(2)
return default
def extract_info_from_html(html_content):
soup = BeautifulSoup(html_content, 'html.parser')
scripts = soup.find_all('script')
token = parent_dir = repo_id = dir_name = None
for script in scripts:
token = extract_var(script.text, 'token', token)
parent_dir = extract_var(script.text, 'path', parent_dir)
repo_id = extract_var(script.text, 'repoID', repo_id)
dir_name = extract_var(script.text, 'dirName', dir_name)
return token, parent_dir, repo_id, dir_name
def get_html_content(url):
response = requests.get(url)
return response.text
def get_upload_url(api_url):
response = requests.get(api_url)
if response.status_code == 200:
return response.json().get('upload_link')
return None
def get_upload_url2(api_url):
headers = {
'Accept': 'application/json',
'X-Requested-With': 'XMLHttpRequest'
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
return response.json().get('url')
return None
def upload_file(upload_url, file_path, fields):
fields = copy.deepcopy(fields)
path = Path(file_path)
filename = path.name
total_size = path.stat().st_size
if not optional_packages:
with open(file_path, 'rb') as f:
fields["file"] = (filename, f)
response = requests.post(upload_url, files=fields, params={'ret-json': 'true'})
return response
# ref: https://stackoverflow.com/a/67726532/11854304
with tqdm(
desc=filename,
total=total_size,
unit="B",
unit_scale=True,
unit_divisor=1024,
) as bar:
with open(file_path, "rb") as f:
fields["file"] = (filename, f)
encoder = MultipartEncoder(fields=fields)
monitor = MultipartEncoderMonitor(
encoder, lambda monitor: bar.update(monitor.bytes_read - bar.n)
)
headers = {"Content-Type": monitor.content_type}
response = requests.post(upload_url, headers=headers, data=monitor, params={'ret-json': 'true'})
return response
def upload_seafile(upload_page_link, file_path_list, verbose):
parsed_results = urlparse(upload_page_link)
base_url = f"{parsed_results.scheme}://{parsed_results.netloc}"
if verbose:
print(f"Input:")
print(f" * Upload page url: {upload_page_link}")
print(f" * Files to be uploaded: {file_path_list}")
print(f"Preparation:")
print(f" * Base url: {base_url}")
# get html content
html_content = get_html_content(upload_page_link)
# extract variables from html content
token, parent_dir, repo_id, dir_name = extract_info_from_html(html_content)
if not parent_dir:
print(f"Cannot extract parent_dir from HTML content.", file=sys.stderr)
return 1
if verbose:
print(f" * dir_name: {dir_name}")
print(f" * parent_dir: {parent_dir}")
# get upload url
upload_url = None
if token:
# ref: https://github.com/haiwen/seafile-js/blob/master/src/seafile-api.js#L1164
api_url = f'{base_url}/api/v2.1/upload-links/{token}/upload/'
upload_url = get_upload_url(api_url)
elif repo_id:
# ref: https://stackoverflow.com/a/38743242/11854304
api_url = upload_page_link.replace("/u/d/", "/ajax/u/d/").rstrip('/') + f"/upload/?r={repo_id}"
upload_url = get_upload_url2(api_url)
if not upload_url:
print(f"Cannot get upload_url.", file=sys.stderr)
return 1
if verbose:
print(f" * upload_url: {upload_url}")
# upload each file
print(f"Upload:")
for idx, file_path in enumerate(file_path_list):
print(f"({idx+1}) {file_path}")
try:
response = upload_file(upload_url, file_path, {'parent_dir': parent_dir})
if response.status_code == 200:
print(f"({idx+1}) upload completed: {response.json()}")
else:
print(f"({idx+1}) {file_path} ERROR: {response.status_code} {response.text}", file=sys.stderr)
except Exception as e:
print(f"({idx+1}) {file_path} EXCEPTION: {e}", file=sys.stderr)
return 0
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('-l', '--link', required=True, help='upload page link (generated by seafile)')
parser.add_argument('-f', '--file', required=True, nargs='+', help='file(s) to upload')
parser.add_argument('--verbose', action='store_true', help='show detailed output')
args = parser.parse_args()
sys.exit(upload_seafile(args.link, args.file, args.verbose))