#!/usr/bin/env python3 """OMX 数据上传站命令行工具（只用 Python 标准库，3.8+）。 文件直接传到阿里云 OSS 深圳（不经过网页服务器），每次上传在站上留记录（谁、何时、路径、大小、sha256、备注）。 云端已有同名文件时一律拒绝，不会覆盖；本工具没有删除功能。 环境变量： OMX_OSS_TOKEN 个人令牌（必填，找管理员要；不要写进代码、不要提交到 git、不要打印出来） OMX_OSS_URL 站点地址，默认 https://omx-oss.pages.dev 用法： python3 omx_upload.py whoami python3 omx_upload.py ls [目录] python3 omx_upload.py upload <本地文件或目录> <云端目标目录> --note "说明" [--batch 批次名] [--dry-run] python3 omx_upload.py records [--days 7] [--mine] python3 omx_upload.py link <云端文件路径> [--days 7] """ import argparse, concurrent.futures as cf, hashlib, json, os, socket, sys, time, urllib.error, urllib.request BASE = os.environ.get("OMX_OSS_URL", "https://omx-oss.pages.dev").rstrip("/") LOG = "omx_upload_log.jsonl" def api(method, path, body=None, query=""): token = os.environ.get("OMX_OSS_TOKEN", "").strip() if not token: sys.exit("缺少环境变量 OMX_OSS_TOKEN（个人令牌，找管理员要）") data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(BASE + "/api/" + path + query, data=data, method=method, headers={"authorization": "Bearer " + token, "content-type": "application/json", "user-agent": "omx-upload-cli/1"}) for attempt in range(4): try: with urllib.request.urlopen(req, timeout=60) as r: return r.status, json.load(r) except urllib.error.HTTPError as e: try: res = e.code, json.load(e) except Exception: res = e.code, {"error": "HTTP %d" % e.code} if e.code >= 500 and attempt < 3: # 站点到 OSS 偶发跨境连接失败，重试 time.sleep(2 * (attempt + 1)) continue return res except (urllib.error.URLError, TimeoutError, ConnectionError) as e: if attempt == 3: raise SystemExit("连不上站点 %s：%s" % (BASE, e)) time.sleep(2 * (attempt + 1)) def put(url, data, headers=None): req = urllib.request.Request(url, data=data, method="PUT", headers=dict(headers or {}, **{"content-length": str(len(data))})) with urllib.request.urlopen(req, timeout=600) as r: return r.status, r.headers.get("ETag", "").strip('"') def sha256_file(p): h = hashlib.sha256() with open(p, "rb") as f: for b in iter(lambda: f.read(1 << 24), b""): h.update(b) return h.hexdigest() def human(n): for u in ("B", "KB", "MB", "GB", "TB"): if n < 1024 or u == "TB": return "%.1f %s" % (n, u) if u != "B" else "%d B" % n n /= 1024 def log(rec): with open(LOG, "a") as f: f.write(json.dumps(rec, ensure_ascii=False) + "\n") def upload_one(local, rel, note, batch, workers): size = os.path.getsize(local) sha = sha256_file(local) st, r = api("POST", "upload/init", {"path": rel, "size": size, "sha256": sha, "note": note, "batch": batch, "source": "agent", "client": socket.gethostname()}) if st == 409: same = (r.get("exists") or {}).get("same") return {"path": rel, "status": "exists", "same_sha256": same, "msg": "云端已有同名文件" + ("（内容相同）" if same else "（内容不同！）")} if st != 200: return {"path": rel, "status": "failed", "msg": r.get("error")} rid = r["rid"] try: if r["mode"] == "single": with open(local, "rb") as f: data = f.read() for attempt in range(4): try: put(r["url"], data, r["headers"]); break except Exception as e: if attempt == 3: raise time.sleep(3) st, c = api("POST", "upload/complete", {"rid": rid, "sha256": sha}) else: ps, n = r["part_size"], r["part_count"] etags, done, cnt = {}, [0], [0] t0 = time.time() def part(i, url): with open(local, "rb") as f: f.seek((i - 1) * ps) data = f.read(ps) for attempt in range(5): try: s, et = put(url, data) if s == 200 and et: done[0] += len(data); cnt[0] += 1 print("\r %s %d/%d 块 %s/s " % (rel, cnt[0], n, human(done[0] / max(1e-3, time.time() - t0))), end="", flush=True) return i, et except Exception: pass if attempt >= 1: # 重签一次，防链接过期 url = api("POST", "upload/sign", {"rid": rid, "parts": [i]})[1]["urls"][str(i)] time.sleep(3 * (attempt + 1)) raise RuntimeError("第 %d 块上传失败" % i) for start in range(1, n + 1, 200): nums = list(range(start, min(n, start + 199) + 1)) st, s = api("POST", "upload/sign", {"rid": rid, "parts": nums}) if st != 200: raise RuntimeError(s.get("error")) with cf.ThreadPoolExecutor(workers) as ex: for i, et in ex.map(lambda k: part(k, s["urls"][str(k)]), nums): etags[i] = et print() st, c = api("POST", "upload/complete", {"rid": rid, "sha256": sha, "parts": [{"n": i, "etag": etags[i]} for i in sorted(etags)]}) if st != 200: raise RuntimeError(c.get("error")) return {"path": rel, "status": "done", "rid": rid, "size": size, "sha256": sha} except BaseException as e: api("POST", "upload/abort", {"rid": rid, "reason": str(e)[:200]}) return {"path": rel, "status": "failed", "rid": rid, "msg": str(e)} def cmd_upload(a): dest = a.dest.strip("/") if os.path.isdir(a.local): files = [] for root, dirs, fs in os.walk(a.local): dirs[:] = sorted(d for d in dirs if not d.startswith(".")) for f in sorted(fs): if not f.startswith("."): p = os.path.join(root, f) files.append((p, dest + "/" + os.path.relpath(p, a.local).replace(os.sep, "/"))) else: files = [(a.local, dest + "/" + os.path.basename(a.local))] total = sum(os.path.getsize(p) for p, _ in files) print("计划上传 %d 个文件，共 %s → %s/%s/" % (len(files), human(total), "omniphysx", dest)) for p, rel in files[:20]: print(" %s → %s (%s)" % (p, rel, human(os.path.getsize(p)))) if len(files) > 20: print(" …… 另有 %d 个" % (len(files) - 20)) if a.dry_run: print("（--dry-run：只列计划，没有上传）") return if not a.note: sys.exit("请用 --note 写清楚这批数据是什么（会记在上传记录里）") res = [] for p, rel in files: r = upload_one(p, rel, a.note, a.batch, a.workers) res.append(r); log(dict(r, local=p, at=time.strftime("%F %T"))) print(" [%s] %s %s" % (r["status"], rel, r.get("msg", ""))) c = {} for r in res: c[r["status"]] = c.get(r["status"], 0) + 1 print("完成：", c, "（明细追加在 %s）" % LOG) if c.get("failed"): sys.exit(1) def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sp = ap.add_subparsers(dest="cmd", required=True) sp.add_parser("whoami") p = sp.add_parser("ls"); p.add_argument("path", nargs="?", default="") p = sp.add_parser("upload"); p.add_argument("local"); p.add_argument("dest") p.add_argument("--note", default=""); p.add_argument("--batch", default=""); p.add_argument("--dry-run", action="store_true") p.add_argument("--workers", type=int, default=4) p = sp.add_parser("records"); p.add_argument("--days", type=int, default=7); p.add_argument("--mine", action="store_true") p = sp.add_parser("link"); p.add_argument("path"); p.add_argument("--days", type=float, default=7) a = ap.parse_args() if a.cmd == "whoami": print(api("GET", "me")[1]) elif a.cmd == "ls": st, r = api("GET", "list", query="?path=" + urllib.request.quote(a.path)) if st != 200: sys.exit(r.get("error")) for f in r["folders"]: print(" [目录] " + f) for f in r["files"]: print(" %12s %s %s" % (human(f["size"]), f["mtime"][:19], f["name"])) elif a.cmd == "upload": cmd_upload(a) elif a.cmd == "records": me = api("GET", "me")[1]["name"] if a.mine else "" st, r = api("GET", "records", query="?days=%d%s" % (a.days, "&user=" + urllib.request.quote(me) if me else "")) for x in r.get("records", []): print("%s %-8s %-6s %-7s %10s %s %s" % (x["started_at"][:19], x["user"], x["source"], x["status"], human(x["size"]), x["path"], x.get("note", "")[:40])) elif a.cmd == "link": st, r = api("GET", "download", query="?path=%s&days=%s" % (urllib.request.quote(a.path), a.days)) print(r.get("url") or r.get("error")) if __name__ == "__main__": main()