Public Access
Initial clean VYTA import
This commit is contained in:
siliconsolder committed
commit
3ea1c3f889
92 files changed
+25789
No files matched your search
@@ -0,0 +1,4 @@
|
||||
.garmin_tokens/
|
||||
garmin-credentials.json
|
||||
*.tmp
|
||||
logs/
|
||||
@@ -0,0 +1,13 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
COPY requirements.txt ./
|
||||
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
CMD ["python", "sync.py"]
|
||||
File diff suppressed because it is too large.
Load diff
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,3 @@
|
||||
garminconnect
|
||||
python-dotenv
|
||||
bleak
|
||||
Executable
+8
@@ -0,0 +1,8 @@
|
||||
@echo off
|
||||
|
||||
cd /d "C:\Users\Silicon Solder\Documents\Codex\vyta-sync"
|
||||
|
||||
"C:\Users\Silicon Solder\Documents\Codex\vyta-sync\.venv\Scripts\python.exe" ^
|
||||
"C:\Users\Silicon Solder\Documents\Codex\vyta-sync\sync.py"
|
||||
|
||||
exit /b %errorlevel%
|
||||
@@ -0,0 +1,288 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
BASE_DIR = os.path.dirname(
|
||||
os.path.abspath(__file__)
|
||||
)
|
||||
|
||||
SYNC_SCRIPT = os.path.join(
|
||||
BASE_DIR,
|
||||
"sync.py",
|
||||
)
|
||||
|
||||
SYNC_INTERVAL = int(
|
||||
os.environ.get(
|
||||
"VYTA_SYNC_INTERVAL_SECONDS",
|
||||
"300",
|
||||
)
|
||||
)
|
||||
|
||||
POLL_INTERVAL = int(
|
||||
os.environ.get(
|
||||
"VYTA_SYNC_POLL_SECONDS",
|
||||
"2",
|
||||
)
|
||||
)
|
||||
|
||||
REQUEST_FILE = os.environ.get(
|
||||
"VYTA_SYNC_REQUEST_PATH",
|
||||
"/data/sync-request.json",
|
||||
)
|
||||
|
||||
STATUS_FILE = os.environ.get(
|
||||
"VYTA_SYNC_STATUS_PATH",
|
||||
"/data/sync-status.json",
|
||||
)
|
||||
|
||||
|
||||
def utc_now():
|
||||
return datetime.now(
|
||||
timezone.utc
|
||||
).isoformat()
|
||||
|
||||
|
||||
def read_status():
|
||||
try:
|
||||
with open(
|
||||
STATUS_FILE,
|
||||
"r",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
loaded = json.load(file)
|
||||
|
||||
return (
|
||||
loaded
|
||||
if isinstance(
|
||||
loaded,
|
||||
dict,
|
||||
)
|
||||
else {}
|
||||
)
|
||||
|
||||
except (
|
||||
FileNotFoundError,
|
||||
json.JSONDecodeError,
|
||||
):
|
||||
return {}
|
||||
|
||||
|
||||
def write_status(
|
||||
**updates,
|
||||
):
|
||||
existing = read_status()
|
||||
|
||||
result = {
|
||||
**existing,
|
||||
**updates,
|
||||
}
|
||||
|
||||
directory = os.path.dirname(
|
||||
STATUS_FILE
|
||||
)
|
||||
|
||||
if directory:
|
||||
os.makedirs(
|
||||
directory,
|
||||
exist_ok=True,
|
||||
)
|
||||
|
||||
temporary_file = (
|
||||
STATUS_FILE +
|
||||
".tmp"
|
||||
)
|
||||
|
||||
with open(
|
||||
temporary_file,
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
json.dump(
|
||||
result,
|
||||
file,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
|
||||
os.replace(
|
||||
temporary_file,
|
||||
STATUS_FILE,
|
||||
)
|
||||
|
||||
|
||||
def consume_manual_request():
|
||||
if not os.path.exists(
|
||||
REQUEST_FILE
|
||||
):
|
||||
return False
|
||||
|
||||
try:
|
||||
os.remove(
|
||||
REQUEST_FILE
|
||||
)
|
||||
|
||||
print(
|
||||
"Manuell sync begärd.",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
|
||||
|
||||
def run_sync(
|
||||
trigger: str,
|
||||
):
|
||||
print(
|
||||
"",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
"=" * 60,
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
f"VYTA sync trigger: {trigger}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
f"Tid: {utc_now()}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
"=" * 60,
|
||||
flush=True,
|
||||
)
|
||||
|
||||
write_status(
|
||||
trigger=trigger,
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
SYNC_SCRIPT,
|
||||
],
|
||||
cwd=BASE_DIR,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
print(
|
||||
"Sync slutade med exit code "
|
||||
f"{result.returncode}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
return result.returncode
|
||||
|
||||
|
||||
def update_countdown(
|
||||
seconds: int,
|
||||
):
|
||||
write_status(
|
||||
nextAutomaticSyncInSeconds=max(
|
||||
0,
|
||||
int(seconds),
|
||||
),
|
||||
syncIntervalSeconds=
|
||||
SYNC_INTERVAL,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
print(
|
||||
"VYTA scheduler startad",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
"Automatisk sync: "
|
||||
f"var {SYNC_INTERVAL} sekund",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
print(
|
||||
"Manuell trigger: "
|
||||
f"{REQUEST_FILE}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
# Kör direkt när containern startar.
|
||||
run_sync(
|
||||
"startup"
|
||||
)
|
||||
|
||||
last_run = time.monotonic()
|
||||
|
||||
while True:
|
||||
now = time.monotonic()
|
||||
|
||||
elapsed = (
|
||||
now -
|
||||
last_run
|
||||
)
|
||||
|
||||
remaining = (
|
||||
SYNC_INTERVAL -
|
||||
elapsed
|
||||
)
|
||||
|
||||
if consume_manual_request():
|
||||
run_sync(
|
||||
"manual"
|
||||
)
|
||||
|
||||
last_run = (
|
||||
time.monotonic()
|
||||
)
|
||||
|
||||
update_countdown(
|
||||
SYNC_INTERVAL
|
||||
)
|
||||
|
||||
time.sleep(
|
||||
POLL_INTERVAL
|
||||
)
|
||||
|
||||
continue
|
||||
|
||||
if remaining <= 0:
|
||||
run_sync(
|
||||
"scheduled"
|
||||
)
|
||||
|
||||
last_run = (
|
||||
time.monotonic()
|
||||
)
|
||||
|
||||
update_countdown(
|
||||
SYNC_INTERVAL
|
||||
)
|
||||
|
||||
time.sleep(
|
||||
POLL_INTERVAL
|
||||
)
|
||||
|
||||
continue
|
||||
|
||||
update_countdown(
|
||||
remaining
|
||||
)
|
||||
|
||||
time.sleep(
|
||||
POLL_INTERVAL
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,379 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
BASE_DIR = os.path.dirname(
|
||||
os.path.abspath(__file__)
|
||||
)
|
||||
|
||||
COLLECTOR = os.path.join(
|
||||
BASE_DIR,
|
||||
"collector.py",
|
||||
)
|
||||
|
||||
NORMALIZER = os.path.join(
|
||||
BASE_DIR,
|
||||
"normalizer.py",
|
||||
)
|
||||
|
||||
UPLOADER = os.path.join(
|
||||
BASE_DIR,
|
||||
"upload_to_blob.py",
|
||||
)
|
||||
|
||||
LOG_DIR = os.path.join(
|
||||
BASE_DIR,
|
||||
"logs",
|
||||
)
|
||||
|
||||
LOG_FILE = os.path.join(
|
||||
LOG_DIR,
|
||||
"sync.log",
|
||||
)
|
||||
|
||||
STATUS_FILE = os.environ.get(
|
||||
"VYTA_SYNC_STATUS_PATH",
|
||||
"/data/sync-status.json",
|
||||
)
|
||||
|
||||
|
||||
def utc_now():
|
||||
return datetime.now(
|
||||
timezone.utc
|
||||
).isoformat()
|
||||
|
||||
|
||||
def log(message: str):
|
||||
os.makedirs(
|
||||
LOG_DIR,
|
||||
exist_ok=True,
|
||||
)
|
||||
|
||||
timestamp = datetime.now().strftime(
|
||||
"%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
line = f"[{timestamp}] {message}"
|
||||
|
||||
print(
|
||||
line,
|
||||
flush=True,
|
||||
)
|
||||
|
||||
with open(
|
||||
LOG_FILE,
|
||||
"a",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
file.write(
|
||||
line + "\n"
|
||||
)
|
||||
|
||||
|
||||
def read_status():
|
||||
try:
|
||||
with open(
|
||||
STATUS_FILE,
|
||||
"r",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
value = json.load(
|
||||
file
|
||||
)
|
||||
|
||||
return (
|
||||
value
|
||||
if isinstance(
|
||||
value,
|
||||
dict,
|
||||
)
|
||||
else {}
|
||||
)
|
||||
|
||||
except (
|
||||
FileNotFoundError,
|
||||
json.JSONDecodeError,
|
||||
):
|
||||
return {}
|
||||
|
||||
|
||||
def write_status(
|
||||
**updates,
|
||||
):
|
||||
existing = read_status()
|
||||
|
||||
result = {
|
||||
**existing,
|
||||
**updates,
|
||||
}
|
||||
|
||||
directory = os.path.dirname(
|
||||
STATUS_FILE
|
||||
)
|
||||
|
||||
if directory:
|
||||
os.makedirs(
|
||||
directory,
|
||||
exist_ok=True,
|
||||
)
|
||||
|
||||
temporary_file = (
|
||||
STATUS_FILE +
|
||||
".tmp"
|
||||
)
|
||||
|
||||
with open(
|
||||
temporary_file,
|
||||
"w",
|
||||
encoding="utf-8",
|
||||
) as file:
|
||||
json.dump(
|
||||
result,
|
||||
file,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
|
||||
os.replace(
|
||||
temporary_file,
|
||||
STATUS_FILE,
|
||||
)
|
||||
|
||||
|
||||
def run_script(
|
||||
script_path: str
|
||||
):
|
||||
name = os.path.basename(
|
||||
script_path
|
||||
)
|
||||
|
||||
log(
|
||||
f"Startar {name}"
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
script_path,
|
||||
],
|
||||
cwd=BASE_DIR,
|
||||
stdin=None,
|
||||
stdout=None,
|
||||
stderr=None,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"{name} misslyckades "
|
||||
f"med exit code "
|
||||
f"{result.returncode}"
|
||||
)
|
||||
|
||||
log(
|
||||
f"{name} klar"
|
||||
)
|
||||
|
||||
|
||||
def should_upload_to_blob():
|
||||
value = os.environ.get(
|
||||
"VYTA_UPLOAD_TO_BLOB",
|
||||
"0",
|
||||
)
|
||||
|
||||
return (
|
||||
value
|
||||
.strip()
|
||||
.lower()
|
||||
in (
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
"on",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
|
||||
def notify_vyta():
|
||||
web_url = os.environ.get(
|
||||
"VYTA_WEB_URL",
|
||||
"",
|
||||
).strip().rstrip("/")
|
||||
|
||||
secret = os.environ.get(
|
||||
"VYTA_INTERNAL_SYNC_SECRET",
|
||||
"",
|
||||
).strip()
|
||||
|
||||
if not web_url:
|
||||
log(
|
||||
"Notifiering hoppas över: "
|
||||
"VYTA_WEB_URL saknas"
|
||||
)
|
||||
return
|
||||
|
||||
if not secret:
|
||||
log(
|
||||
"Notifiering hoppas över: "
|
||||
"VYTA_INTERNAL_SYNC_SECRET saknas"
|
||||
)
|
||||
return
|
||||
|
||||
url = (
|
||||
web_url +
|
||||
"/api/notifications/evaluate"
|
||||
)
|
||||
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
data=b"{}",
|
||||
method="POST",
|
||||
headers={
|
||||
"Authorization":
|
||||
f"Bearer {secret}",
|
||||
"Content-Type":
|
||||
"application/json",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
request,
|
||||
timeout=15,
|
||||
) as response:
|
||||
body = (
|
||||
response
|
||||
.read()
|
||||
.decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
)
|
||||
|
||||
log(
|
||||
"Notifieringsutvärdering klar "
|
||||
f"(HTTP {response.status}): "
|
||||
f"{body}"
|
||||
)
|
||||
|
||||
except urllib.error.HTTPError as error:
|
||||
body = (
|
||||
error
|
||||
.read()
|
||||
.decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
)
|
||||
|
||||
log(
|
||||
"VARNING: "
|
||||
"Notifieringsutvärdering "
|
||||
f"gav HTTP {error.code}: "
|
||||
f"{body}"
|
||||
)
|
||||
|
||||
except Exception as error:
|
||||
log(
|
||||
"VARNING: "
|
||||
"Notifieringsutvärdering "
|
||||
f"misslyckades: {error}"
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
started_at = utc_now()
|
||||
|
||||
write_status(
|
||||
status="syncing",
|
||||
startedAt=started_at,
|
||||
error=None,
|
||||
)
|
||||
|
||||
log(
|
||||
"=" * 60
|
||||
)
|
||||
|
||||
log(
|
||||
"VYTA SYNC START"
|
||||
)
|
||||
|
||||
try:
|
||||
run_script(
|
||||
COLLECTOR
|
||||
)
|
||||
|
||||
run_script(
|
||||
NORMALIZER
|
||||
)
|
||||
|
||||
if should_upload_to_blob():
|
||||
run_script(
|
||||
UPLOADER
|
||||
)
|
||||
else:
|
||||
log(
|
||||
"Hoppar över Vercel Blob-upload"
|
||||
)
|
||||
|
||||
notify_vyta()
|
||||
|
||||
completed_at = utc_now()
|
||||
|
||||
write_status(
|
||||
status="idle",
|
||||
startedAt=started_at,
|
||||
completedAt=completed_at,
|
||||
lastSuccessfulSync=completed_at,
|
||||
error=None,
|
||||
)
|
||||
|
||||
log(
|
||||
"VYTA SYNC KLAR"
|
||||
)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
write_status(
|
||||
status="idle",
|
||||
error="Sync avbruten",
|
||||
)
|
||||
|
||||
log(
|
||||
"VYTA SYNC AVBRUTEN"
|
||||
)
|
||||
|
||||
sys.exit(
|
||||
130
|
||||
)
|
||||
|
||||
except Exception as error:
|
||||
write_status(
|
||||
status="error",
|
||||
completedAt=utc_now(),
|
||||
error=str(
|
||||
error
|
||||
),
|
||||
)
|
||||
|
||||
log(
|
||||
f"SYNC MISSLYCKADES: "
|
||||
f"{error}"
|
||||
)
|
||||
|
||||
sys.exit(
|
||||
1
|
||||
)
|
||||
|
||||
finally:
|
||||
log(
|
||||
"=" * 60
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,70 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import date
|
||||
from getpass import getpass
|
||||
|
||||
from garminconnect import Garmin
|
||||
|
||||
|
||||
TOKENSTORE = os.path.join(os.path.dirname(__file__), ".garmin_tokens")
|
||||
|
||||
|
||||
def dump(title, data):
|
||||
print()
|
||||
print("=" * 70)
|
||||
print(title)
|
||||
print("=" * 70)
|
||||
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
|
||||
|
||||
|
||||
def main():
|
||||
email = input("Garmin email: ").strip()
|
||||
password = getpass("Garmin password: ")
|
||||
|
||||
garmin = Garmin(
|
||||
email=email,
|
||||
password=password,
|
||||
)
|
||||
|
||||
os.makedirs(TOKENSTORE, exist_ok=True)
|
||||
|
||||
print("\nLoggar in mot Garmin Connect...")
|
||||
|
||||
result = garmin.login(TOKENSTORE)
|
||||
|
||||
print("Inloggning OK.")
|
||||
print("Display name:", garmin.display_name)
|
||||
|
||||
today = date.today().isoformat()
|
||||
|
||||
print(f"\nHämtar Garmin-data för {today}...")
|
||||
|
||||
calls = [
|
||||
("Daily stats", lambda: garmin.get_stats(today)),
|
||||
("Heart rate", lambda: garmin.get_heart_rates(today)),
|
||||
("Sleep", lambda: garmin.get_sleep_data(today)),
|
||||
("HRV", lambda: garmin.get_hrv_data(today)),
|
||||
("Stress", lambda: garmin.get_stress_data(today)),
|
||||
("Body Battery", lambda: garmin.get_body_battery(today)),
|
||||
("Respiration", lambda: garmin.get_respiration_data(today)),
|
||||
("SpO2", lambda: garmin.get_spo2_data(today)),
|
||||
("Training readiness", lambda: garmin.get_training_readiness(today)),
|
||||
("Training status", lambda: garmin.get_training_status(today)),
|
||||
("Activities", lambda: garmin.get_activities(0, 10)),
|
||||
("Devices", lambda: garmin.get_devices()),
|
||||
]
|
||||
|
||||
for title, fn in calls:
|
||||
try:
|
||||
data = fn()
|
||||
dump(title, data)
|
||||
except Exception as e:
|
||||
print()
|
||||
print("=" * 70)
|
||||
print(title)
|
||||
print("=" * 70)
|
||||
print(f"ERROR: {type(e).__name__}: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,166 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# CONFIG
|
||||
# ------------------------------------------------------------
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
|
||||
VYTA_PROJECT = Path(
|
||||
r"C:\Users\Silicon Solder\Documents\Codex\vyta"
|
||||
)
|
||||
|
||||
DATA_FILE = (
|
||||
VYTA_PROJECT
|
||||
/ "private-data"
|
||||
/ "vyta-data.json"
|
||||
)
|
||||
|
||||
BLOB_PATH = "vyta/vyta-data.json"
|
||||
|
||||
|
||||
# Läs vyta-sync\.env
|
||||
load_dotenv(SCRIPT_DIR / ".env")
|
||||
|
||||
|
||||
# ------------------------------------------------------------
|
||||
# HELPERS
|
||||
# ------------------------------------------------------------
|
||||
|
||||
def find_vercel_cli() -> str:
|
||||
"""
|
||||
Hitta Vercel CLI.
|
||||
|
||||
På Windows ligger global npm CLI normalt som vercel.cmd.
|
||||
"""
|
||||
|
||||
candidates = [
|
||||
"vercel.cmd",
|
||||
"vercel",
|
||||
]
|
||||
|
||||
for candidate in candidates:
|
||||
executable = shutil.which(candidate)
|
||||
|
||||
if executable:
|
||||
return executable
|
||||
|
||||
# Vanlig npm global-bin på Windows
|
||||
appdata = os.environ.get("APPDATA")
|
||||
|
||||
if appdata:
|
||||
fallback = Path(appdata) / "npm" / "vercel.cmd"
|
||||
|
||||
if fallback.exists():
|
||||
return str(fallback)
|
||||
|
||||
raise FileNotFoundError(
|
||||
"Kunde inte hitta Vercel CLI. "
|
||||
"Kontrollera att 'vercel --version' fungerar i PowerShell."
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("VYTA → VERCEL PRIVATE BLOB")
|
||||
print("=" * 60)
|
||||
|
||||
# --------------------------------------------------------
|
||||
# Kontrollera datafil
|
||||
# --------------------------------------------------------
|
||||
|
||||
if not DATA_FILE.exists():
|
||||
print()
|
||||
print("ERROR: Vyta-datafilen saknas:")
|
||||
print(DATA_FILE)
|
||||
sys.exit(1)
|
||||
|
||||
print()
|
||||
print(f"Datafil: {DATA_FILE}")
|
||||
print(f"Storlek: {DATA_FILE.stat().st_size:,} bytes")
|
||||
|
||||
# --------------------------------------------------------
|
||||
# Blob-token
|
||||
# --------------------------------------------------------
|
||||
|
||||
token = os.environ.get(
|
||||
"BLOB_READ_WRITE_TOKEN"
|
||||
)
|
||||
|
||||
if not token:
|
||||
print()
|
||||
print(
|
||||
"ERROR: BLOB_READ_WRITE_TOKEN saknas i vyta-sync\\.env"
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
print("Blob-token: OK")
|
||||
|
||||
# --------------------------------------------------------
|
||||
# Vercel CLI
|
||||
# --------------------------------------------------------
|
||||
|
||||
try:
|
||||
vercel_cli = find_vercel_cli()
|
||||
except FileNotFoundError as exc:
|
||||
print()
|
||||
print(f"ERROR: {exc}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Vercel CLI: {vercel_cli}")
|
||||
|
||||
# --------------------------------------------------------
|
||||
# Upload
|
||||
# --------------------------------------------------------
|
||||
|
||||
command = [
|
||||
vercel_cli,
|
||||
"blob",
|
||||
"put",
|
||||
str(DATA_FILE),
|
||||
"--pathname",
|
||||
BLOB_PATH,
|
||||
"--access",
|
||||
"private",
|
||||
"--allow-overwrite",
|
||||
"--content-type",
|
||||
"application/json",
|
||||
"--rw-token",
|
||||
token,
|
||||
]
|
||||
|
||||
print()
|
||||
print(
|
||||
f"Laddar upp till Private Blob: {BLOB_PATH}"
|
||||
)
|
||||
print()
|
||||
|
||||
result = subprocess.run(
|
||||
command,
|
||||
cwd=str(VYTA_PROJECT),
|
||||
text=True,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
print()
|
||||
print(
|
||||
f"ERROR: Blob-upload misslyckades "
|
||||
f"(exit code {result.returncode})"
|
||||
)
|
||||
sys.exit(result.returncode)
|
||||
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("VYTA BLOB-UPLOAD KLAR")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user