C:\TripServer\backups\admin_server\admin_server_backup_20260715_151945.py
# TripServer update compatibility marker.
# The currently installed legacy uploader scans only the first 400,000 characters.
# Required identity tokens: Flask @app.route app.run BASE_DIR ADMIN_PORT restart_process_later upload_admin_server run_self_test_suite build_watchdog_code watchdog_status
# -*- coding: utf-8 -*-
from flask import Flask, request, redirect, send_from_directory, make_response, Response, send_file
from werkzeug.utils import secure_filename
import os
import time
import sys
import re
import shutil
import py_compile
import threading
import html as html_lib
import glob
import zipfile
import subprocess
import ipaddress
import json
import urllib.request
import urllib.parse
import uuid
import hashlib
import math
import struct
import mimetypes
import tempfile
from pathlib import Path
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
app = Flask(__name__)
ADMIN_PANEL_VERSION = "TripServer Admin v55 Enforced FX Kids Budget App"
ADMIN_STARTED_AT = time.strftime("%Y-%m-%d %H:%M:%S")
BASE_DIR = r"C:\TripServer"
ADMIN_HOST = "0.0.0.0"
ADMIN_PORT = 8000
CONTROL_PORT = 8010
CONTROL_DIR = os.path.join(BASE_DIR, "control_server")
CONTROL_SCRIPT_PATH = os.path.join(CONTROL_DIR, "tripserver_control_server.py")
CONTROL_PENDING_ADMIN_PATH = os.path.join(CONTROL_DIR, "pending_admin_server.py")
CONTROL_REQUEST_PATH = os.path.join(CONTROL_DIR, "request.json")
CONTROL_STATUS_PATH = os.path.join(CONTROL_DIR, "status.json")
CONTROL_LOG_PATH = os.path.join(CONTROL_DIR, "tripserver_control_server.log")
CONTROL_BOOTSTRAP_LOG_PATH = os.path.join(BASE_DIR, "tripserver_control_bootstrap.log")
CONTROL_TASK_NAME = "TripServerControlServer"
ADMIN_FAILSAFE_STATUS_PATH = os.path.join(BASE_DIR, "admin_failsafe_status.json")
PRESENTATION_MUSIC_UPLOAD_STATUS_PATH = os.path.join(BASE_DIR, "presentation_music_upload_status.json")
PENDING_MUSIC_UPLOAD_DIR = os.path.join(BASE_DIR, "tmp", "pending_music_uploads")
# Local File Manager roots are discovered at runtime on the Windows server.
# Access is intentionally limited to discovered Downloads folders and TripServer folders.
LOCAL_FILE_ROOTS_CACHE = None
LOCAL_FILE_ROOTS_CACHE_TS = 0
LOCAL_FILE_ALLOWED_ROOT_NAMES = {"downloads", "tripserver"}
LOCAL_FILE_TEXT_EXTENSIONS = {".txt", ".log", ".json", ".md", ".py", ".html", ".htm", ".css", ".js", ".csv", ".bat", ".ps1", ".xml"}
LOCAL_FILE_PREVIEW_MAX_BYTES = 1024 * 1024
LOCAL_FILE_ZIP_MAX_BYTES = 120 * 1024 * 1024
LOCAL_FILE_LIST_LIMIT = 700
FOLDERS = {
"JAPAN": os.path.join(BASE_DIR, "JAPAN"),
"USA": os.path.join(BASE_DIR, "USA"),
"USA_TV": os.path.join(BASE_DIR, "USA_TV"),
"Moneyapp": os.path.join(BASE_DIR, "Moneyapp"),
"MoneyHome": os.path.join(BASE_DIR, "MoneyHome"),
"TripForm": os.path.join(BASE_DIR, "TripForm"),
}
APP_FOLDERS = {"JAPAN", "USA", "USA_TV", "Moneyapp", "MoneyHome", "TripForm"}
# Preferred entry file for each app root. This prevents /USA_TV/ from opening a presentation by accident
# when the folder contains more than one HTML file.
APP_ENTRY_FILES = {
"JAPAN": ["index.html"],
"USA": ["index.html"],
"USA_TV": ["usa2027-tv.html", "index.html"],
"Moneyapp": ["index.html"],
"MoneyHome": ["index.html"],
"TripForm": ["index.html"],
}
# Netlify export builder.
# Uses the same current trip files that the admin panel serves from C:\TripServer\JAPAN and C:\TripServer\USA.
# The generated ZIP is rewritten only for deployment paths: budget app URL and presentation launch URL.
NETLIFY_BUDGET_APP_URL = "https://beautiful-kitsune-dafd66.netlify.app/"
NETLIFY_EXPORT_DIR = os.path.join(BASE_DIR, "tmp", "netlify_exports")
# Canonical USA Netlify-only budget slide image.
# Extracted from the approved manual Netlify bundle; it contains no overlaid budget amount.
# It is written only into the temporary Netlify build and never replaces the private-server image.
# Canonical USA Netlify-only budget slide HTML.
# The background image is shared; the no-amount behavior is defined by this slide markup, not by the image.
USA_NETLIFY_BUDGET_SLIDE_HTML = '<section class="slide budget" data-key="budget">\n<div class="slide-shell">\n<div class="media-panel">\n<img alt="תקציב הטיול" class="hero-image" data-local-base="media/usa-presentation/images/budget" loading="lazy" referrerpolicy="no-referrer" src="media/usa-presentation/images/budget.jpg"/>\n<div class="image-overlay"></div>\n<div class="image-caption">תכנון תקציבי</div>\n</div>\n<div class="info-panel">\n<div class="badge">💰 תקציב</div>\n<h1>תקציב — בלי דרמה</h1>\n<div class="region">תמונת מצב תקציבית</div>\n<p class="vibe">השקף הזה משאיר את המספרים בקובץ הטיול, ובמצגת מציג רק את התמונה הכללית.</p>\n<ul class="summary-list"><li>המסגרת התקציבית מנוהלת בקובץ הטיול ובאפליקציית התקציב.</li><li>המעקב מחולק ללינה, פארקים, נסיעות, אוכל, קניות וקרוז.</li><li>המטרה: ליהנות מהטיול בלי להפוך כל עצירה לחישוב.</li></ul>\n<div class="fun-note">את המספרים המדויקים משאירים למעקב התקציב — כאן נשארים באווירה של טיול.</div>\n</div>\n</div>\n</section>'
NETLIFY_BUILD_DIR = os.path.join(BASE_DIR, "tmp", "netlify_build")
NETLIFY_STATIC_EXTENSIONS = {
".html", ".htm", ".css", ".js", ".json", ".txt", ".md", ".svg", ".xml", ".map",
".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp", ".ico",
".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".opus", ".wma", ".webm", ".mp4", ".m4v",
".woff", ".woff2", ".ttf"
}
NETLIFY_SKIP_DIRS = {"__pycache__", ".git", ".svn", ".hg", "backups", "backup", "tmp", "temp", "node_modules"}
NETLIFY_SKIP_FILES = {"thumbs.db", ".ds_store", "desktop.ini"}
NETLIFY_CONTROL_FILES = {"_redirects", "_headers"}
# USA Netlify is the phone/computer deployment. TV files live under USA_TV and must never
# leak into the USA deploy bundle. A stale TV presentation inside C:\TripServer\USA used
# to make validation fail because it referenced images-tv paths that are intentionally absent.
NETLIFY_USA_EXCLUDED_FILES = {
"usa2027-presentation-tv.html",
"usa2027-tv.html",
}
NETLIFY_USA_EXCLUDED_PREFIXES = {
"media/usa-presentation/images-tv/",
}
JAPAN_PRESENTATION_FILES = {
"flight_intro": "japan-flight-intro.html",
"japan_experience": "japan-experience.html",
}
USA_PRESENTATION_FILES = {
"usa_presentation": "usa2027-presentation.html",
"usa_presentation_tv": "usa2027-presentation-tv.html",
}
USA_EXTRA_HTML_FILES = {
"usa_trip_tv": "usa2027-tv.html",
}
USA_PRESENTATION_ROOTS = {
"usa_presentation": "USA",
"usa_presentation_tv": "USA_TV",
}
USA_EXTRA_HTML_ROOTS = {
"usa_trip_tv": "USA_TV",
}
USA_MUSIC_FILES = {
"usa_music": os.path.join("media", "usa-presentation", "music", "usa-roadtrip-theme.mp3"),
"usa_music_part2": os.path.join("media", "usa-presentation", "music", "usa-part2-theme.mp3"),
}
USA_IMAGE_KEYS = ({"intro", "budget", "roadtrip-map", "summary", "extra-big-sur", "extra-diner", "extra-death-valley", "extra-mono-lake"} | {f"day-{i:02d}" for i in range(1, 37)})
USA_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}
USA_VIDEO_EXTENSIONS = {".mp4", ".m4v"}
USA_MEDIA_EXTENSIONS = USA_IMAGE_EXTENSIONS | USA_VIDEO_EXTENSIONS
USA_MEDIA_MANIFEST_NAME = "media-manifest.json"
USA_MEDIA_TIMELINE_REPORT_NAME = "media-timeline-report.json"
USA_MEDIA_PATCH_VERSION = "2026-07-13-v44-inline-source-of-truth"
USA_VIDEO_MAX_SINGLE_FILE_BYTES = int(os.environ.get("TRIPSERVER_USA_VIDEO_MAX_BYTES", str(220 * 1024 * 1024)))
USA_MEDIA_ZIP_MAX_BYTES = int(os.environ.get("TRIPSERVER_USA_MEDIA_ZIP_MAX_BYTES", str(750 * 1024 * 1024)))
USA_MEDIA_ZIP_MAX_UNCOMPRESSED_BYTES = int(os.environ.get("TRIPSERVER_USA_MEDIA_ZIP_MAX_UNCOMPRESSED_BYTES", str(1500 * 1024 * 1024)))
USA_MEDIA_ZIP_MAX_SINGLE_FILE_BYTES = int(os.environ.get("TRIPSERVER_USA_MEDIA_ZIP_MAX_SINGLE_FILE_BYTES", str(240 * 1024 * 1024)))
USA_MEDIA_ZIP_MAX_FILES = 220
USA_IMAGE_TARGETS = {
"phone": {
"label": "מסך פלאפון",
"root_key": "USA",
"description": "מצגת ארה״ב למסכי אייפון/אנדרואיד",
"folder": os.path.join("media", "usa-presentation", "images"),
"rel_prefix": "media/usa-presentation/images",
"presentation_key": "usa_presentation",
"backup_tag": "USA_images_phone",
},
"tv": {
"label": "מסך טלוויזיה",
"description": "מצגת ארה״ב למסך TV / LG C3 77 4K",
"root_key": "USA_TV",
"folder": os.path.join("media", "usa-presentation", "images"),
"rel_prefix": "media/usa-presentation/images",
"presentation_key": "usa_presentation_tv",
"backup_tag": "USA_TV_images",
},
}
JAPAN_MUSIC_TARGETS = {
# The Japan experience presentation contains several fallback paths. Keep all aliases synced.
"japan_experience_music": [
os.path.join("media", "japan-experience", "music", "japan-theme.mp3"),
os.path.join("media", "music", "japan-theme.mp3"),
"japan-theme.mp3",
],
"japan_flight_music": [
os.path.join("media", "japan-flight", "music", "flight-intro.mp3"),
],
}
# Presentation music sync rules.
# v34.14 keeps existing music upload routes as the single source of truth.
# Music buttons in the UI now upload the file, measure it, and synchronize the relevant presentation section.
# Japan main song syncs all japan-experience slides. USA part 1/2 sync only their own PART_*_SECONDS values.
PRESENTATION_MUSIC_SYNC_RULES = {
"japan_experience_music": {
"label": "יפן קיץ 26 — שיר מצגת ראשית",
"trip": "JAPAN",
"presentation": "japan-experience.html",
"music_targets": JAPAN_MUSIC_TARGETS["japan_experience_music"],
"scope": "כל שקפי מצגת יפן",
"future_behavior": "פעיל כעת: העלאת שיר מצגת יפן מודדת את האורך לפני הטמעה. אם ממוצע השקף יוצא מתחת ל־5 שניות או מעל ל־10 שניות, מוצג מסך אישור לפני שמירת השיר וסנכרון המצגת.",
"enabled_now": True,
},
"japan_flight_music": {
"label": "יפן קיץ 26 — מוזיקת פתיח טיסה",
"trip": "JAPAN",
"presentation": "japan-flight-intro.html",
"music_targets": JAPAN_MUSIC_TARGETS["japan_flight_music"],
"scope": "פתיח הטיסה בלבד",
"future_behavior": "העלאת שיר תישמר לפתיח הטיסה בלבד; לא מסנכרנת את מצגת יפן הראשית.",
"enabled_now": False,
},
"usa_music_part1": {
"label": "ארצות הברית קיץ 27 — שיר חלק 1",
"trip": "USA",
"presentation": "usa2027-presentation.html / usa2027-presentation-tv.html",
"music_targets": [os.path.join("USA", USA_MUSIC_FILES["usa_music"]), os.path.join("USA_TV", USA_MUSIC_FILES["usa_music"])],
"scope": "שקפים 1–23 בלבד",
"future_behavior": "פעיל כעת: העלאת שיר חלק 1 מסנכרנת רק את מקטע 1 ולא משנה את מקטע 2.",
"enabled_now": True,
},
"usa_music_part2": {
"label": "ארצות הברית קיץ 27 — שיר חלק 2",
"trip": "USA",
"presentation": "usa2027-presentation.html / usa2027-presentation-tv.html",
"music_targets": [os.path.join("USA", USA_MUSIC_FILES["usa_music_part2"]), os.path.join("USA_TV", USA_MUSIC_FILES["usa_music_part2"])],
"scope": "שקפים 24–44 בלבד",
"future_behavior": "פעיל כעת: העלאת שיר חלק 2 מסנכרנת רק את מקטע 2 ולא משנה את מקטע 1.",
"enabled_now": True,
},
}
# Japan main presentation music timing guardrails.
# The uploaded file is first converted into a temporary pending MP3, measured, and only then embedded.
# If the detected song length would create an average slide duration outside this range, the admin asks for confirmation.
JAPAN_EXPERIENCE_MIN_AVG_SLIDE_SECONDS = 5.0
JAPAN_EXPERIENCE_MAX_AVG_SLIDE_SECONDS = 10.0
JAPAN_EXPERIENCE_LOOP_MIN_SLIDE_SECONDS = 5.0
PENDING_MUSIC_MAX_AGE_SECONDS = 60 * 60
AUDIO_UPLOAD_EXTENSIONS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".opus", ".wma", ".webm", ".mp4"}
JAPAN_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff"}
JAPAN_IMAGES_ZIP_MAX_BYTES = 130 * 1024 * 1024
JAPAN_IMAGES_ZIP_MAX_FILES = 220
JAPAN_IMAGES_ZIP_MAX_UNCOMPRESSED_BYTES = 220 * 1024 * 1024
JAPAN_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES = 18 * 1024 * 1024
TRIP_PACKAGE_ZIP_MAX_BYTES = 180 * 1024 * 1024
TRIP_PACKAGE_ZIP_MAX_FILES = 320
TRIP_PACKAGE_ZIP_MAX_UNCOMPRESSED_BYTES = 260 * 1024 * 1024
TRIP_PACKAGE_ZIP_MAX_SINGLE_FILE_BYTES = 35 * 1024 * 1024
TRIP_PACKAGE_EXTENSIONS = {".html", ".htm", ".jpg", ".jpeg", ".png", ".webp", ".mp3", ".mp4", ".m4v", ".css", ".js", ".json", ".txt", ".svg"}
ADMIN_SERVER_PATH = os.path.abspath(__file__)
ADMIN_BACKUP_DIR = os.path.join(BASE_DIR, "backups", "admin_server")
ADMIN_UPLOAD_TMP_DIR = os.path.join(BASE_DIR, "tmp", "admin_server_uploads")
ADMIN_MAX_UPLOAD_BYTES = 2 * 1024 * 1024
ADMIN_MAX_PACKAGE_UPLOAD_BYTES = 8 * 1024 * 1024
USA_IMAGES_ZIP_MAX_BYTES = 80 * 1024 * 1024
USA_IMAGES_ZIP_MAX_FILES = 160
USA_IMAGES_ZIP_MAX_UNCOMPRESSED_BYTES = 140 * 1024 * 1024
USA_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES = 12 * 1024 * 1024
ADMIN_RESTART_CMD_PATH = os.path.join(BASE_DIR, "restart_admin_server.cmd")
ADMIN_RESTART_LOG_PATH = os.path.join(BASE_DIR, "admin_restart.log")
ADMIN_RESTART_HELPER_PATH = os.path.join(BASE_DIR, "restart_admin_server_helper.py")
UPLOAD_BACKUP_DIR = os.path.join(BASE_DIR, "backups", "uploaded_files")
# External watchdog. This is intentionally separate from the Flask process: if the admin
# server is down, Windows Task Scheduler can still run this script every minute and bring
# the admin back without walking to the physical computer.
WATCHDOG_TASK_NAME = os.environ.get("TRIPSERVER_WATCHDOG_TASK", "TripServerAdminWatchdog")
WATCHDOG_SCRIPT_PATH = os.path.join(BASE_DIR, "tripserver_admin_watchdog.py")
WATCHDOG_LOG_PATH = os.path.join(BASE_DIR, "tripserver_watchdog.log")
WATCHDOG_STATE_PATH = os.path.join(BASE_DIR, "tripserver_watchdog_state.txt")
WATCHDOG_INTERVAL_MINUTES = int(os.environ.get("TRIPSERVER_WATCHDOG_INTERVAL_MINUTES", "5"))
WATCHDOG_MIN_RESTART_INTERVAL_SECONDS = int(os.environ.get("TRIPSERVER_WATCHDOG_RESTART_COOLDOWN", "90"))
# v27: self-update/restart is launched through Windows Task Scheduler, not as a
# direct child of the running Flask process. This prevents the old admin process
# from killing the helper that is supposed to bring the new admin back online.
RESTART_TASK_NAME = os.environ.get("TRIPSERVER_RESTART_TASK", "TripServerAdminRestartOnce")
def get_pythonw_executable(preferred=None):
"""Return a no-console Python executable for Windows scheduled tasks when available.
Running the watchdog every few minutes with python.exe can flash a black console
window and steal focus. pythonw.exe runs the same script without a console.
On non-Windows or if pythonw.exe is not available, fall back to the normal
interpreter.
"""
exe = preferred or sys.executable or "python"
if os.name != "nt":
return exe
try:
base = os.path.basename(exe).lower()
folder = os.path.dirname(exe)
if base == "pythonw.exe" and os.path.exists(exe):
return exe
if folder:
candidate = os.path.join(folder, "pythonw.exe")
if os.path.exists(candidate):
return candidate
# Last-resort common launcher name. Do not validate here because it may be in PATH.
return "pythonw.exe"
except Exception:
return exe
# Admin log viewer. Only these fixed paths are exposed in the UI; user-supplied paths are never accepted.
LOG_FILE_REGISTRY = {
"restart": {"label": "לוג אתחול פאנל", "path": ADMIN_RESTART_LOG_PATH, "note": "Restart helper, העלאת admin חדש, שחזור גיבוי ופעולות מחשב."},
"control": {"label": "לוג שרת בקרה", "path": CONTROL_LOG_PATH, "note": "שרת הבקרה החיצוני שמפעיל ומעדכן את האדמין הראשי."},
"control_bootstrap": {"label": "לוג התקנת שרת בקרה", "path": CONTROL_BOOTSTRAP_LOG_PATH, "note": "כתיבת והפעלת שרת הבקרה."},
"watchdog": {"label": "לוג legacy", "path": WATCHDOG_LOG_PATH, "note": "שאריות לוג ממנגנונים ישנים, אם קיימות. לא חלק מהפאנל הראשי."},
}
LOG_TAIL_DEFAULT_LINES = 120
LOG_TAIL_MAX_LINES = 500
# Smart media optimizer settings.
# Images uploaded to fixed slide slots are always stored as real .jpg files.
# Audio uploaded to fixed music slots is always stored as .mp3 when FFmpeg is available.
IMAGE_OPTIMIZER_MAX_LONG_EDGE = int(os.environ.get("TRIPSERVER_IMAGE_MAX_LONG_EDGE", "1920"))
IMAGE_OPTIMIZER_TARGET_BYTES = int(os.environ.get("TRIPSERVER_IMAGE_TARGET_BYTES", str(1100 * 1024)))
IMAGE_OPTIMIZER_QUALITIES = tuple(int(q) for q in os.environ.get("TRIPSERVER_IMAGE_QUALITIES", "88,86,84,82,80,78").split(",") if q.strip())
AUDIO_UPLOAD_MAX_BYTES = int(os.environ.get("TRIPSERVER_AUDIO_MAX_BYTES", str(80 * 1024 * 1024)))
AUDIO_MP3_BITRATE = os.environ.get("TRIPSERVER_AUDIO_MP3_BITRATE", "192k")
AUDIO_MP3_SAMPLE_RATE = os.environ.get("TRIPSERVER_AUDIO_MP3_SAMPLE_RATE", "44100")
MEDIA_AUDIO_ACCEPT = ".mp3,.wav,.m4a,.aac,.flac,.ogg,.opus,.wma,.webm,.mp4,audio/*,video/mp4"
MEDIA_IMAGE_ACCEPT = ".jpg,.jpeg,.png,.webp,.bmp,.tif,.tiff,image/*"
# Power actions are intentionally guarded: POST only, LAN/local client only, typed Hebrew confirmation, and cancel support.
POWER_ACTIONS_ENABLED = os.environ.get("TRIPSERVER_ENABLE_POWER_ACTIONS", "1").strip().lower() not in {"0", "false", "no", "off"}
POWER_MIN_DELAY_SECONDS = 30
POWER_MAX_DELAY_SECONDS = 3600
POWER_DEFAULT_DELAY_SECONDS = 60
POWER_CONFIRM_TEXT = {
"shutdown": "כיבוי",
"reboot": "הפעלה מחדש",
}
POWER_LABELS = {
"shutdown": "כיבוי מחשב",
"reboot": "ריסטארט למחשב",
}
ADMIN_UPDATE_FAILSAFE_RESTART_SECONDS = int(os.environ.get("TRIPSERVER_ADMIN_UPDATE_FAILSAFE_RESTART_SECONDS", "180"))
ADMIN_UPDATE_FAILSAFE_ENABLED = os.environ.get("TRIPSERVER_ADMIN_UPDATE_FAILSAFE", "1").strip().lower() not in {"0", "false", "no", "off"}
def _json_safe_write(path, payload):
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2)
os.replace(tmp, path)
return True
except Exception:
return False
def read_admin_failsafe_status():
try:
if os.path.exists(ADMIN_FAILSAFE_STATUS_PATH):
with open(ADMIN_FAILSAFE_STATUS_PATH, "r", encoding="utf-8") as fh:
data = json.load(fh)
if isinstance(data, dict):
return data
except Exception:
pass
return {"state": "unknown", "message": "עדיין אין נתוני Fail Safe שמורים."}
def write_admin_failsafe_status(state, message, **extra):
payload = {
"state": state,
"message": message,
"updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"admin_version": ADMIN_PANEL_VERSION,
}
payload.update(extra)
_json_safe_write(ADMIN_FAILSAFE_STATUS_PATH, payload)
return payload
def is_no_pending_shutdown_message(output):
txt = (output or "").lower()
return ("1116" in txt) or ("no shutdown was in progress" in txt) or ("אין" in txt and "כיבוי" in txt)
def render_failsafe_status_card():
data = read_admin_failsafe_status()
state = str(data.get("state") or "unknown")
message = str(data.get("message") or "")
updated = str(data.get("updated_at") or "")
if state in {"armed"}:
icon = "🟠"
title = "Fail Safe חמוש"
elif state in {"cancelled", "auto_cancelled", "none"}:
icon = "🟢"
title = "Fail Safe לא פעיל"
elif state in {"arm_failed", "cancel_failed"}:
icon = "🔴"
title = "Fail Safe דורש בדיקה"
else:
icon = "⚪"
title = "Fail Safe ללא סטטוס"
safe_msg = html_lib.escape(message or "אין ריסטארט פעיל כרגע או שעדיין אין בדיקת סטטוס.")
safe_updated = html_lib.escape(updated)
return '<div class="status-card"><b>%s %s</b><span>%s<br><small>%s</small></span></div>' % (icon, title, safe_msg, safe_updated)
# Webcam / room camera control. The camera is never opened automatically on page load.
# It starts only when the user presses the camera button, and it is guarded to LAN/Tailscale/private IPs.
CAMERA_ENABLED = os.environ.get("TRIPSERVER_ENABLE_WEBCAM", "1").strip().lower() not in {"0", "false", "no", "off"}
CAMERA_INDEX = int(os.environ.get("TRIPSERVER_WEBCAM_INDEX", "0"))
CAMERA_WIDTH = int(os.environ.get("TRIPSERVER_WEBCAM_WIDTH", "1280"))
CAMERA_HEIGHT = int(os.environ.get("TRIPSERVER_WEBCAM_HEIGHT", "720"))
CAMERA_FPS = max(1, min(20, int(os.environ.get("TRIPSERVER_WEBCAM_FPS", "8"))))
CAMERA_JPEG_QUALITY = max(45, min(92, int(os.environ.get("TRIPSERVER_WEBCAM_JPEG_QUALITY", "78"))))
CAMERA_MAX_STREAM_SECONDS = int(os.environ.get("TRIPSERVER_WEBCAM_MAX_STREAM_SECONDS", "1800"))
CAMERA_FRAME_INTERVAL_SECONDS = 1.0 / float(CAMERA_FPS)
# Optional audio + recording layer. Audio and recording are manual-only and guarded by the same
# LAN/Tailscale/private-IP checks as the live camera. FFmpeg is required for audio and MP4 recording.
CAMERA_AUDIO_ENABLED = os.environ.get("TRIPSERVER_ENABLE_CAMERA_AUDIO", "1").strip().lower() not in {"0", "false", "no", "off"}
CAMERA_RECORDING_ENABLED = os.environ.get("TRIPSERVER_ENABLE_CAMERA_RECORDING", "1").strip().lower() not in {"0", "false", "no", "off"}
CAMERA_VIDEO_DEVICE = os.environ.get("TRIPSERVER_FFMPEG_VIDEO_DEVICE", "").strip()
CAMERA_AUDIO_DEVICE = os.environ.get("TRIPSERVER_FFMPEG_AUDIO_DEVICE", "").strip()
CAMERA_RECORDINGS_DIR = os.environ.get("TRIPSERVER_CAMERA_RECORDINGS_DIR", os.path.join(BASE_DIR, "camera_recordings"))
CAMERA_RECORD_FPS = max(1, min(30, int(os.environ.get("TRIPSERVER_CAMERA_RECORD_FPS", "15"))))
CAMERA_RECORD_MAX_SECONDS = max(30, min(8 * 3600, int(os.environ.get("TRIPSERVER_CAMERA_RECORD_MAX_SECONDS", "3600"))))
CAMERA_AUDIO_BITRATE = os.environ.get("TRIPSERVER_CAMERA_AUDIO_BITRATE", "128k")
# Camera component installer. First-run auto installer: if OpenCV is missing, the admin
# starts a background pip installation automatically when the Flask process starts.
# It uses the same Python executable that runs this admin server and writes a log.
# The camera itself still never turns on automatically; only the dependency installer runs.
CAMERA_SETUP_LOG_PATH = os.path.join(BASE_DIR, "camera_setup.log")
CAMERA_AUDIO_LOG_PATH = os.path.join(BASE_DIR, "camera_audio_stream.log")
CAMERA_HLS_DIR = os.path.join(BASE_DIR, "camera_live_hls")
CAMERA_HLS_LOG_PATH = os.path.join(BASE_DIR, "camera_hls_stream.log")
CAMERA_SETUP_CONFIRM_TEXT = "התקן"
CAMERA_OPENCV_PACKAGE = os.environ.get("TRIPSERVER_OPENCV_PACKAGE", "opencv-python").strip() or "opencv-python"
CAMERA_OPENCV_INSTALL_TIMEOUT_SECONDS = int(os.environ.get("TRIPSERVER_OPENCV_INSTALL_TIMEOUT_SECONDS", "900"))
CAMERA_AUTO_INSTALL_OPENCV = os.environ.get("TRIPSERVER_CAMERA_AUTO_INSTALL_OPENCV", "1").strip().lower() not in {"0", "false", "no", "off"}
CAMERA_AUTO_INSTALL_DELAY_SECONDS = int(os.environ.get("TRIPSERVER_CAMERA_AUTO_INSTALL_DELAY_SECONDS", "4"))
CAMERA_SETUP_MARKER_PATH = os.path.join(BASE_DIR, "camera_opencv_installed.ok")
CAMERA_SETUP_STATE = {"running": False, "started_at": "", "finished_at": "", "ok": False, "last_message": "", "mode": "idle", "auto_requested": False}
CAMERA_SETUP_LOCK = threading.RLock()
LOG_FILE_REGISTRY["camera_setup"] = {"label": "לוג התקנת רכיבי מצלמה", "path": CAMERA_SETUP_LOG_PATH, "note": "התקנת OpenCV דרך pip ובדיקת רכיבי מצלמה."}
LOG_FILE_REGISTRY["camera_audio"] = {"label": "לוג אודיו מצלמה", "path": CAMERA_AUDIO_LOG_PATH, "note": "FFmpeg DirectShow, בחירת מיקרופון וניסיונות הפעלת אודיו."}
LOG_FILE_REGISTRY["camera_hls"] = {"label": "לוג שידור מצלמה מאוחד HLS", "path": CAMERA_HLS_LOG_PATH, "note": "FFmpeg DirectShow וידאו+אודיו בתהליך אחד, כדי למנוע נעילת התקני USB."}
def safe_zip_relpath(raw_name):
raw = (raw_name or "").replace("\\", "/").strip("/")
if not raw or raw.startswith("__MACOSX/"):
return None
parts = [p for p in raw.split("/") if p and p not in (".", "..")]
if not parts or len(parts) != len(raw.split("/")):
return None
return "/".join(parts)
def file_ext(filename):
return os.path.splitext(filename or "")[1].lower()
def is_likely_jpeg_file(path):
try:
with open(path, "rb") as fh:
return fh.read(3) == b"\xff\xd8\xff"
except Exception:
return False
def is_likely_mp3_file(path):
"""Fast MP3 sniffing: ID3 tag or MPEG audio frame sync near the beginning."""
try:
with open(path, "rb") as fh:
data = fh.read(4096)
if data.startswith(b"ID3"):
return True
for idx in range(0, max(0, len(data) - 1)):
if data[idx] == 0xFF and (data[idx + 1] & 0xE0) == 0xE0:
return True
return False
except Exception:
return False
def find_ffmpeg_executable():
candidates = [
os.environ.get("FFMPEG_PATH"),
"ffmpeg",
r"C:\ffmpeg\bin\ffmpeg.exe",
r"C:\Program Files\ffmpeg\bin\ffmpeg.exe",
r"C:\Program Files (x86)\ffmpeg\bin\ffmpeg.exe",
]
for candidate in candidates:
if not candidate:
continue
found = shutil.which(candidate) if os.path.basename(candidate) == candidate else (candidate if os.path.exists(candidate) else None)
if found:
return found
return None
def find_ffprobe_executable():
candidates = [
os.environ.get("FFPROBE_PATH"),
"ffprobe",
r"C:\ffmpeg\bin\ffprobe.exe",
r"C:\Program Files\ffmpeg\bin\ffprobe.exe",
r"C:\Program Files (x86)\ffmpeg\bin\ffprobe.exe",
]
for candidate in candidates:
if not candidate:
continue
found = shutil.which(candidate) if os.path.basename(candidate) == candidate else (candidate if os.path.exists(candidate) else None)
if found:
return found
return None
def run_hidden_subprocess(cmd, timeout=20):
kwargs = dict(capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, shell=False)
if os.name == "nt":
kwargs["creationflags"] = getattr(subprocess, "CREATE_NO_WINDOW", 0)
return subprocess.run(cmd, **kwargs)
def format_seconds(value):
try:
total = int(round(float(value)))
minutes, seconds = divmod(total, 60)
hours, minutes = divmod(minutes, 60)
if hours:
return f"{hours}:{minutes:02d}:{seconds:02d}"
return f"{minutes}:{seconds:02d}"
except Exception:
return str(value)
def probe_audio_duration_seconds(path):
ffprobe = find_ffprobe_executable()
if not ffprobe:
return {"ok": False, "error": r"ffprobe לא נמצא. ודא שקיים C:\ffmpeg\bin\ffprobe.exe", "ffprobe": ""}
cmd = [
ffprobe, "-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
path,
]
try:
cp = run_hidden_subprocess(cmd, timeout=20)
raw = ((cp.stdout or "") + "\n" + (cp.stderr or "")).strip()
if cp.returncode != 0:
return {"ok": False, "error": raw or f"ffprobe failed rc={cp.returncode}", "return_code": cp.returncode, "ffprobe": ffprobe}
value = None
for line in raw.splitlines():
line = line.strip()
if not line:
continue
try:
value = float(line)
break
except Exception:
pass
if value is None or value <= 0:
return {"ok": False, "error": "ffprobe לא החזיר משך שיר תקין", "raw": raw, "ffprobe": ffprobe}
return {"ok": True, "duration_seconds": value, "duration_display": format_seconds(value), "raw": raw, "ffprobe": ffprobe}
except subprocess.TimeoutExpired:
return {"ok": False, "error": "ffprobe חרג מזמן הבדיקה", "ffprobe": ffprobe}
except Exception as exc:
return {"ok": False, "error": repr(exc), "ffprobe": ffprobe}
def music_key_to_rule_key(trip, music_key):
trip = (trip or "").upper()
if trip == "JAPAN":
return music_key if music_key in PRESENTATION_MUSIC_SYNC_RULES else None
if trip == "USA":
if music_key == "usa_music":
return "usa_music_part1"
if music_key == "usa_music_part2":
return "usa_music_part2"
return None
def usa_folder_for_presentation_key(presentation_key):
root_key = USA_PRESENTATION_ROOTS.get(presentation_key, "USA")
return FOLDERS.get(root_key, FOLDERS["USA"])
def usa_folder_for_extra_html_key(extra_key):
root_key = USA_EXTRA_HTML_ROOTS.get(extra_key, "USA")
return FOLDERS.get(root_key, FOLDERS["USA"])
def usa_folder_for_image_target(target_variant):
target_variant, info = usa_image_target_info(target_variant)
return FOLDERS.get(info.get("root_key") or "USA", FOLDERS["USA"])
def usa_music_absolute_targets(rel_path):
# USA phone and USA_TV use the same relative media path inside different roots.
return [
os.path.join(FOLDERS["USA"], rel_path),
os.path.join(FOLDERS["USA_TV"], rel_path),
]
def save_audio_upload_to_absolute_targets(uploaded_file, target_paths, backup_group):
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_name = secure_filename(uploaded_file.filename or "audio_upload") or "audio_upload"
source_tmp = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"audio_abs_{stamp}_{os.getpid()}_{safe_name}")
converted_tmp = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"audio_abs_{stamp}_{os.getpid()}_converted.mp3")
uploaded_file.save(source_tmp)
try:
media = convert_or_copy_audio_to_mp3(source_tmp, converted_tmp, uploaded_file.filename or "")
sha = file_sha256(converted_tmp)
results = []
for target_path in target_paths:
ensure_parent_dir(target_path)
backup_path = backup_existing_file(target_path, backup_group)
tmp_target = target_path + ".tmp"
shutil.copy2(converted_tmp, tmp_target)
os.replace(tmp_target, target_path)
results.append({"rel": os.path.relpath(target_path, BASE_DIR), "path": target_path, "size": os.path.getsize(target_path), "sha256": file_sha256(target_path), "backup": backup_path})
return {"size": os.path.getsize(converted_tmp), "sha256": sha, "targets": results, "media": media}
finally:
for p in (source_tmp, converted_tmp):
try:
if os.path.exists(p):
os.remove(p)
except Exception:
pass
def read_presentation_music_upload_status():
try:
if os.path.exists(PRESENTATION_MUSIC_UPLOAD_STATUS_PATH):
with open(PRESENTATION_MUSIC_UPLOAD_STATUS_PATH, "r", encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except Exception:
return {}
return {}
def write_presentation_music_upload_status(rule_key, status):
data = read_presentation_music_upload_status()
data[rule_key or "unknown"] = status
history = data.get("_history")
if not isinstance(history, list):
history = []
history.insert(0, status)
data["_history"] = history[:30]
ensure_parent_dir(PRESENTATION_MUSIC_UPLOAD_STATUS_PATH)
tmp = PRESENTATION_MUSIC_UPLOAD_STATUS_PATH + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
os.replace(tmp, PRESENTATION_MUSIC_UPLOAD_STATUS_PATH)
def primary_saved_audio_path(save_result):
if not isinstance(save_result, dict):
return None
if save_result.get("path"):
return save_result.get("path")
targets = save_result.get("targets")
if isinstance(targets, list) and targets:
first = targets[0]
if isinstance(first, dict):
return first.get("path")
return None
def all_saved_audio_paths(save_result):
if not isinstance(save_result, dict):
return []
if save_result.get("path"):
return [save_result.get("path")]
targets = save_result.get("targets")
if isinstance(targets, list):
return [item.get("path") for item in targets if isinstance(item, dict) and item.get("path")]
return []
def read_text_file_utf8(path):
with open(path, "r", encoding="utf-8", errors="ignore") as f:
return f.read()
def write_text_file_utf8(path, new_text):
"""Write UTF-8 text atomically without creating a second backup.
Callers that already created a transaction backup use this helper so a path
correction cannot fail halfway through or leave a truncated presentation.
"""
ensure_parent_dir(path)
tmp_path = path + ".tmp"
try:
with open(tmp_path, "w", encoding="utf-8", newline="") as f:
f.write(new_text)
f.flush()
try:
os.fsync(f.fileno())
except OSError:
pass
os.replace(tmp_path, path)
except Exception:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
raise
def write_text_file_utf8_atomic_with_backup(path, new_text, backup_group):
if not os.path.exists(path):
raise FileNotFoundError(path)
backup_path = backup_existing_file(path, backup_group)
tmp_path = path + ".tmp"
with open(tmp_path, "w", encoding="utf-8", newline="") as f:
f.write(new_text)
os.replace(tmp_path, path)
return backup_path
def calculate_scaled_slide_durations_ms(old_durations, target_total_ms):
return calculate_scaled_slide_durations_ms_with_min(old_durations, target_total_ms, None)
def calculate_scaled_slide_durations_ms_with_min(old_durations, target_total_ms, minimum_per_slide_ms=None):
if not old_durations:
raise ValueError("לא נמצאו זמני שקפים לחישוב.")
target_total_ms = int(round(float(target_total_ms)))
if target_total_ms <= 0:
raise ValueError("משך יעד לא תקין לסנכרון שקפים.")
current_total = sum(int(x) for x in old_durations)
if current_total <= 0:
raise ValueError("סך זמני השקפים הנוכחי אינו תקין.")
count = len(old_durations)
if minimum_per_slide_ms is None:
minimum = 1200 if target_total_ms >= count * 1200 else max(250, target_total_ms // max(1, count))
else:
minimum = int(round(float(minimum_per_slide_ms)))
if minimum <= 0:
raise ValueError("רצפת זמן שקף אינה תקינה.")
if target_total_ms < count * minimum:
raise ValueError("משך היעד קצר מדי ביחס לרצפת זמן השקפים שנבחרה.")
scale = target_total_ms / float(current_total)
new_durations = [max(minimum, int(round(int(x) * scale))) for x in old_durations]
diff = target_total_ms - sum(new_durations)
if diff > 0:
new_durations[-1] += diff
elif diff < 0:
remaining = -diff
for i in range(len(new_durations) - 1, -1, -1):
room = max(0, new_durations[i] - minimum)
if room <= 0:
continue
take = min(room, remaining)
new_durations[i] -= take
remaining -= take
if remaining <= 0:
break
if remaining > 0:
if minimum_per_slide_ms is None:
new_durations[-1] = max(1, new_durations[-1] - remaining)
else:
raise ValueError("לא ניתן להגיע למשך היעד בלי לרדת מתחת לרצפת זמן השקף.")
if sum(new_durations) != target_total_ms:
new_durations[-1] += target_total_ms - sum(new_durations)
if any(int(x) <= 0 for x in new_durations):
raise ValueError("חישוב זמני השקפים יצר ערך לא תקין.")
if minimum_per_slide_ms is not None and any(int(x) < int(round(float(minimum_per_slide_ms))) for x in new_durations):
raise ValueError("חישוב זמני השקפים ירד מתחת לרצפת זמן השקף שנבחרה.")
return new_durations
def get_japan_experience_slide_durations_ms():
path = os.path.join(FOLDERS["JAPAN"], JAPAN_PRESENTATION_FILES["japan_experience"])
if not os.path.exists(path):
raise FileNotFoundError(path)
html = read_text_file_utf8(path)
match = re.search(r"const\s+SLIDES\s*=\s*(\[.*?\]);", html, flags=re.S)
if not match:
raise ValueError("לא נמצא מערך SLIDES במצגת יפן. אסור לנחש זמני שקפים.")
block = match.group(1)
old_durations = [int(m.group(1)) for m in re.finditer(r'"duration"\s*:\s*(\d+)', block)]
if not old_durations:
raise ValueError("לא נמצאו ערכי duration במערך SLIDES של מצגת יפן.")
return {"path": path, "html": html, "match": match, "block": block, "durations": old_durations}
def sync_japan_experience_presentation_to_target_duration(duration_seconds, sync_mode="matched_music_length", source_duration_seconds=None, minimum_per_slide_seconds=None):
data = get_japan_experience_slide_durations_ms()
html = data["html"]
match = data["match"]
block = data["block"]
old_durations = data["durations"]
target_total_ms = int(round(float(duration_seconds) * 1000))
min_ms = None if minimum_per_slide_seconds is None else int(round(float(minimum_per_slide_seconds) * 1000))
new_durations = calculate_scaled_slide_durations_ms_with_min(old_durations, target_total_ms, min_ms)
index = {"value": 0}
def replace_duration(m):
i = index["value"]
index["value"] += 1
return '"duration":' + str(new_durations[i])
new_block = re.sub(r'"duration"\s*:\s*\d+', replace_duration, block)
if index["value"] != len(new_durations):
raise ValueError("מספר ערכי duration שהוחלפו אינו תואם למספר השקפים.")
source_text = ""
if source_duration_seconds is not None:
source_text = f" source song: {float(source_duration_seconds):.6f}s;"
floor_text = ""
if minimum_per_slide_seconds is not None:
floor_text = f" min slide: {float(minimum_per_slide_seconds):.3f}s;"
stamp = (
f"// TripServer music sync v34.20 confirm: japan-experience.html; mode: {sync_mode};"
f"{source_text} target total: {float(duration_seconds):.6f}s; slides: {len(new_durations)};"
f" old total: {sum(old_durations)/1000:.3f}s; new total: {sum(new_durations)/1000:.3f}s;{floor_text}"
)
new_html = html[:match.start(1)] + new_block + html[match.end(1):]
if "// TripServer music sync v34.20 confirm: japan-experience.html" in new_html:
new_html = re.sub(r"// TripServer music sync v34\.20 confirm: japan-experience\.html[^\n]*\n", stamp + "\n", new_html, count=1)
elif "// TripServer music sync v34.19 safe: japan-experience.html" in new_html:
new_html = re.sub(r"// TripServer music sync v34\.19 safe: japan-experience\.html[^\n]*\n", stamp + "\n", new_html, count=1)
elif "// TripServer music sync v34.14: japan-experience.html" in new_html:
new_html = re.sub(r"// TripServer music sync v34\.14: japan-experience\.html[^\n]*\n", stamp + "\n", new_html, count=1)
elif "// TripServer music sync:" in new_html:
new_html = re.sub(r"// TripServer music sync:[^\n]*\n", stamp + "\n", new_html, count=1)
else:
new_html = new_html[:match.start()] + stamp + "\n" + new_html[match.start():]
if new_html == html:
raise ValueError("לא בוצע שינוי בפועל במצגת יפן.")
backup_path = write_text_file_utf8_atomic_with_backup(data["path"], new_html, "JAPAN_presentation_music_sync")
verify = get_japan_experience_slide_durations_ms()
verify_durations = verify.get("durations") or []
if len(verify_durations) != len(new_durations) or sum(verify_durations) != sum(new_durations):
raise ValueError("אימות לאחר כתיבה נכשל: זמני השקפים בקובץ אינם תואמים לחישוב.")
return {
"ok": True,
"path": data["path"],
"backup": backup_path,
"sha256": file_sha256(data["path"]),
"slides": len(new_durations),
"old_total_seconds": sum(old_durations) / 1000.0,
"new_total_seconds": sum(new_durations) / 1000.0,
"duration_seconds": float(duration_seconds),
"source_duration_seconds": None if source_duration_seconds is None else float(source_duration_seconds),
"min_slide_seconds": None if minimum_per_slide_seconds is None else float(minimum_per_slide_seconds),
"average_slide_seconds": (sum(new_durations) / 1000.0) / max(1, len(new_durations)),
"sync_mode": sync_mode,
"message": f"מצגת יפן עודכנה: {len(new_durations)} שקפים, סך חדש {format_seconds(sum(new_durations)/1000.0)}.",
}
def sync_japan_experience_presentation_to_music_duration(duration_seconds):
"""Synchronize japan-experience.html exactly to the uploaded song duration.
v34.20 keeps this exact-sync behavior only after the user explicitly approves it
or when the measured average slide time is already inside the safe 5–10 second range.
"""
result = sync_japan_experience_presentation_to_target_duration(
duration_seconds,
sync_mode="matched_music_length",
source_duration_seconds=duration_seconds,
minimum_per_slide_seconds=None,
)
result["message"] = f"מצגת יפן סונכרנה לפי שיר באורך {format_seconds(duration_seconds)} ({float(duration_seconds):.3f} שניות)."
return result
def sync_usa_presentation_part_to_music_duration(music_key, duration_seconds):
"""Synchronize one USA song part transactionally across phone and TV.
The two-part music structure remains unchanged:
- PART_1_SECONDS controls slides 1-23.
- PART_2_SECONDS controls slides 24-44.
Before either presentation is changed, the prospective timeline is calculated with
all currently installed videos. Videos keep their full duration and image slides
receive the remaining song time equally. If the new song is too short, nothing is
written. If a later write/verification step fails, every changed presentation is
restored to its exact previous contents.
"""
key = music_key_to_rule_key("USA", music_key) or music_key
if key == "usa_music_part1":
const_name = "PART_1_SECONDS"
label = "ארה״ב חלק 1"
scope = "שקפים 1–23"
elif key == "usa_music_part2":
const_name = "PART_2_SECONDS"
label = "ארה״ב חלק 2"
scope = "שקפים 24–44"
else:
raise ValueError("סוג שיר ארה״ב לא מוכר לסנכרון.")
seconds = float(duration_seconds)
if not math.isfinite(seconds) or seconds <= 0:
raise ValueError("משך שיר לא תקין לסנכרון ארה״ב.")
seconds_text = f"{seconds:.6f}"
candidates = []
# Phase 1: build and validate both prospective files without changing disk.
for presentation_key in ("usa_presentation", "usa_presentation_tv"):
filename = USA_PRESENTATION_FILES.get(presentation_key)
if not filename:
continue
path = os.path.join(usa_folder_for_presentation_key(presentation_key), filename)
if not os.path.exists(path):
continue
html = read_text_file_utf8(path)
pattern = re.compile(r"(const\s+" + re.escape(const_name) + r"\s*=\s*)([0-9]+(?:\.[0-9]+)?)(\s*;)")
match = pattern.search(html)
if not match:
raise ValueError(f"לא נמצא {const_name} בתוך {filename}. אסור לנחש זמני מצגת.")
old_value = float(match.group(2))
new_html = pattern.sub(r"\g<1>" + seconds_text + r"\g<3>", html, count=1)
if const_name == "PART_1_SECONDS":
new_html = re.sub(r"//\s*שקפים\s*1-23:[^\n]*", f"// שקפים 1-23: usa-roadtrip-theme, {seconds_text} שניות.", new_html, count=1)
else:
new_html = re.sub(r"//\s*שקפים\s*24-44:[^\n]*", f"// שקפים 24-44: usa-part2-theme, {seconds_text} שניות.", new_html, count=1)
stamp = f"// TripServer music sync v34.33: {label} ({scope}) סונכרן לפי שיר באורך {seconds_text} שניות."
same_part_stamp = re.compile(r"// TripServer music sync v(?:34\.14|34\.33): " + re.escape(label) + r"[^\n]*\n")
if same_part_stamp.search(new_html):
new_html = same_part_stamp.sub(stamp + "\n", new_html, count=1)
else:
# Keep the exact position stable: insert next to the timing constant.
refreshed_match = pattern.search(new_html)
insert_at = refreshed_match.start() if refreshed_match else match.start()
new_html = new_html[:insert_at] + stamp + "\n" + new_html[insert_at:]
if new_html == html:
raise ValueError(f"לא בוצע שינוי בפועל ב־{filename}.")
target_variant = "tv" if presentation_key == "usa_presentation_tv" else "phone"
prospective_info, prospective_timeline, prospective_videos = calculate_usa_timeline_for_html(target_variant, new_html)
candidates.append({
"presentation_key": presentation_key,
"target_variant": target_variant,
"filename": filename,
"path": path,
"old_html": html,
"new_html": new_html,
"old_seconds": old_value,
"prospective_timeline": prospective_timeline,
"prospective_video_count": len(prospective_videos),
})
if not candidates:
raise FileNotFoundError("לא נמצאה מצגת ארה״ב לסנכרון: usa2027-presentation.html או usa2027-presentation-tv.html")
# Phase 2: write all files, then regenerate/verify their media metadata.
written = []
results = []
try:
for item in candidates:
backup_path = write_text_file_utf8_atomic_with_backup(item["path"], item["new_html"], "USA_presentation_music_sync")
item["backup"] = backup_path
written.append(item)
for item in candidates:
media_refresh = refresh_usa_media_after_timing_change(item["target_variant"])
results.append({
"ok": True,
"path": item["path"],
"backup": item.get("backup"),
"sha256": file_sha256(item["path"]),
"media_timeline_refresh": media_refresh,
"presentation_key": item["presentation_key"],
"filename": item["filename"],
"part": key,
"const_name": const_name,
"scope": scope,
"old_seconds": item["old_seconds"],
"new_seconds": seconds,
"duration_seconds": seconds,
"video_count": item["prospective_video_count"],
"timeline_total_seconds": item["prospective_timeline"].total_seconds,
"message": f"{item['filename']}: {const_name} עודכן מ־{item['old_seconds']:.6f} ל־{seconds_text} שניות.",
})
except Exception:
# Exact rollback of every presentation touched in this transaction.
for item in reversed(written):
try:
tmp = item["path"] + ".music_sync_rollback.tmp"
with open(tmp, "w", encoding="utf-8", newline="") as fh:
fh.write(item["old_html"])
os.replace(tmp, item["path"])
except Exception:
pass
for item in candidates:
try:
write_usa_media_metadata(item["target_variant"])
except Exception:
pass
raise
return {
"ok": True,
"part": key,
"label": label,
"scope": scope,
"duration_seconds": seconds,
"results": results,
"message": f"{label} סונכרן לפי שיר באורך {format_seconds(seconds)} ({seconds:.3f} שניות).",
}
def rollback_saved_audio_upload(save_result):
"""Restore every audio target from its pre-upload backup after sync failure."""
restored = []
failed = []
if not isinstance(save_result, dict):
return {"ok": False, "restored": restored, "failed": ["save_result לא תקין"]}
targets = save_result.get("targets") or []
for item in targets:
if not isinstance(item, dict) or not item.get("path"):
continue
path = item.get("path")
backup = item.get("backup")
try:
if backup and os.path.isfile(backup):
ensure_parent_dir(path)
tmp = path + ".audio_rollback.tmp"
shutil.copy2(backup, tmp)
os.replace(tmp, path)
restored.append({"path": path, "from": backup, "action": "restored_backup"})
else:
if os.path.exists(path):
os.remove(path)
restored.append({"path": path, "from": "", "action": "removed_new_file"})
except Exception as exc:
failed.append({"path": path, "error": repr(exc)})
return {"ok": not failed, "restored": restored, "failed": failed}
def register_existing_music_upload_duration(trip, music_key, save_result):
rule_key = music_key_to_rule_key(trip, music_key)
rule = PRESENTATION_MUSIC_SYNC_RULES.get(rule_key or "", {})
path = primary_saved_audio_path(save_result)
paths = all_saved_audio_paths(save_result)
status = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"trip": (trip or "").upper(),
"music_key": music_key,
"rule_key": rule_key or "",
"label": rule.get("label", ""),
"scope": rule.get("scope", ""),
"presentation": rule.get("presentation", ""),
"paths": paths,
"primary_path": path or "",
"duration_ok": False,
"duration_seconds": None,
"duration_display": "",
"sync_performed": False,
"sync_note": "v34.14: העלאת שיר קיימת מודדת את האורך ומסנכרנת את המצגת/המקטע הרלוונטיים כאשר כלל הסנכרון פעיל.",
}
if not path or not os.path.exists(path):
status["error"] = "קובץ השיר נשמר אך לא נמצא נתיב קובץ למדידת אורך."
if rule_key:
write_presentation_music_upload_status(rule_key, status)
return status
probe = probe_audio_duration_seconds(path)
status["ffprobe"] = probe.get("ffprobe", "")
if probe.get("ok"):
status["duration_ok"] = True
status["duration_seconds"] = probe.get("duration_seconds")
status["duration_display"] = probe.get("duration_display")
try:
sync_result = None
if (trip or "").upper() == "JAPAN" and music_key == "japan_experience_music":
sync_result = sync_japan_experience_presentation_to_music_duration(probe.get("duration_seconds"))
status["sync_note"] = "בוצע סנכרון בפועל: כל שקפי מצגת יפן סונכרנו לפי אורך השיר החדש."
elif (trip or "").upper() == "USA" and rule_key in {"usa_music_part1", "usa_music_part2"}:
sync_result = sync_usa_presentation_part_to_music_duration(rule_key, probe.get("duration_seconds"))
status["sync_note"] = "בוצע סנכרון בפועל: מקטע השיר הרלוונטי עודכן גם במצגת ארה״ב הרגילה וגם במצגת TV כאשר הקבצים קיימים."
elif (trip or "").upper() == "JAPAN" and music_key == "japan_flight_music":
status["sync_note"] = "מוזיקת פתיח הטיסה נשמרה ונמדדה. לא שונה japan-experience.html כי פתיח הטיסה הוא מקטע נפרד."
if sync_result:
status["sync_performed"] = True
if isinstance(sync_result, dict) and sync_result.get("results"):
status["sync_results"] = sync_result.get("results")
status["sync_summary"] = sync_result.get("message", "")
else:
status["sync_results"] = [sync_result]
status["sync_summary"] = sync_result.get("message", "") if isinstance(sync_result, dict) else ""
except Exception as sync_exc:
status["sync_performed"] = False
status["sync_error"] = repr(sync_exc)
if (trip or "").upper() == "USA":
rollback = rollback_saved_audio_upload(save_result)
status["audio_rollback"] = rollback
status["audio_rollback_performed"] = bool(rollback.get("ok"))
if rollback.get("ok"):
status["sync_note"] = "סנכרון המצגת נכשל ולכן קובצי המוזיקה החדשים בוטלו ושוחזרו אוטומטית לגרסה הקודמת."
else:
status["sync_note"] = "סנכרון המצגת נכשל. בוצע ניסיון שחזור מוזיקה, אך לפחות יעד אחד דורש בדיקה ידנית."
else:
status["sync_note"] = "השיר נשמר ונמדד, אך סנכרון המצגת נכשל. אין אישור שהמצגת מסונכרנת."
else:
status["error"] = probe.get("error") or probe.get("raw") or "ffprobe לא הצליח לזהות משך שיר."
if rule_key:
write_presentation_music_upload_status(rule_key, status)
return status
def music_upload_duration_details(status):
if not isinstance(status, dict):
return ""
lines = ["סטטוס אורך שיר:"]
if status.get("label"):
lines.append(f"שיר/מקטע: {status.get('label')}")
if status.get("scope"):
lines.append(f"תחום סנכרון עתידי: {status.get('scope')}")
if status.get("duration_ok"):
lines.append(f"משך שזוהה: {status.get('duration_display')} ({float(status.get('duration_seconds') or 0):.3f} שניות)")
if status.get("sync_performed"):
lines.append(f"סנכרון שקפים: בוצע בפועל עבור {status.get('label') or 'המצגת'}.")
for item in status.get("sync_results") or []:
if isinstance(item, dict):
lines.append(f"מצגת: {item.get('path','')}")
if item.get('const_name'):
lines.append(f"מקטע: {item.get('scope','')} | קבוע שעודכן: {item.get('const_name')} | סך חדש: {float(item.get('new_seconds') or item.get('duration_seconds') or 0):.3f} שניות")
else:
lines.append(f"שקפים: {item.get('slides','')} | סך חדש: {float(item.get('new_total_seconds') or 0):.3f} שניות")
if item.get("backup"):
lines.append(f"גיבוי מצגת קודם: {item.get('backup')}")
elif status.get("sync_error"):
lines.append("סנכרון שקפים: נכשל.")
lines.append(f"שגיאת סנכרון: {status.get('sync_error')}")
if status.get("audio_rollback_performed"):
lines.append("שחזור מוזיקה: בוצע בהצלחה; קובצי השיר הקודמים הוחזרו.")
elif status.get("audio_rollback"):
lines.append("שחזור מוזיקה: לא הושלם במלואו; יש לבדוק את פירוט השחזור.")
lines.append(json.dumps(status.get("audio_rollback"), ensure_ascii=False, indent=2))
else:
lines.append("סנכרון שקפים: לא בוצע בשלב הזה.")
else:
lines.append("אזהרה: השיר נשמר, אבל לא זוהה משך תקין ולכן אין אישור לסנכרון.")
if status.get("error"):
lines.append(f"שגיאה: {status.get('error')}")
if status.get("ffprobe"):
lines.append(f"ffprobe: {status.get('ffprobe')}")
if status.get("primary_path"):
lines.append(f"נתיב שנבדק: {status.get('primary_path')}")
lines.append("קובץ סטטוס: " + PRESENTATION_MUSIC_UPLOAD_STATUS_PATH)
return "\n".join(lines)
def render_music_upload_status_rows():
data = read_presentation_music_upload_status()
rows = []
for key, rule in PRESENTATION_MUSIC_SYNC_RULES.items():
st = data.get(key) if isinstance(data, dict) else None
if isinstance(st, dict) and st.get("timestamp"):
if st.get("duration_ok"):
if st.get("sync_performed"):
badge = f"<span class='badge safe'>סונכרן: {html_lib.escape(str(st.get('duration_display') or ''))}</span>"
detail = f"משך זוהה וסנכרון בוצע בפועל עבור {html_lib.escape(str(st.get('label') or rule.get('label') or 'המצגת'))}."
elif st.get("sync_error"):
badge = f"<span class='badge warn'>נמדד, סנכרון נכשל</span>"
detail = html_lib.escape(str(st.get("sync_error") or "שגיאת סנכרון"))
else:
badge = f"<span class='badge safe'>נמדד: {html_lib.escape(str(st.get('duration_display') or ''))}</span>"
detail = "משך זוהה. עדיין לא בוצע סנכרון שקפים."
else:
badge = "<span class='badge warn'>שגיאת מדידה</span>"
detail = html_lib.escape(str(st.get("error") or "לא זוהה משך"))
timestamp = html_lib.escape(str(st.get("timestamp") or ""))
else:
badge = "<span class='badge warn'>אין העלאה אחרונה</span>"
detail = "טרם נרשם סטטוס העלאה לשיר הזה."
timestamp = "—"
rows.append(f"<tr><td><b>{html_lib.escape(rule.get('label',''))}</b></td><td>{badge}</td><td>{timestamp}</td><td>{detail}</td></tr>")
return "".join(rows)
def media_engine_status():
ffmpeg_path = find_ffmpeg_executable() or ""
ffprobe_path = find_ffprobe_executable() or ""
status = {"pillow": False, "ffmpeg": bool(ffmpeg_path), "ffmpeg_path": ffmpeg_path, "ffprobe": bool(ffprobe_path), "ffprobe_path": ffprobe_path}
try:
import PIL # noqa: F401
status["pillow"] = True
except Exception:
status["pillow"] = False
return status
def format_bytes(num):
try:
num = int(num)
except Exception:
return str(num)
units = ["B", "KB", "MB", "GB"]
value = float(num)
for unit in units:
if value < 1024 or unit == units[-1]:
return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} B"
value /= 1024
def convert_or_copy_image_to_jpg(source_path, target_path, original_name=""):
"""Create a real optimized JPG file; never writes a fake .jpg.
Smart behavior:
- Uses Pillow when installed.
- Applies EXIF orientation.
- Converts alpha/transparency to a white background.
- Downscales only when the long edge is larger than IMAGE_OPTIMIZER_MAX_LONG_EDGE.
- Uses progressive/optimized JPEG and a quality loop to reduce weight without aggressive damage.
- If Pillow is missing, accepts only true JPEG bytes and copies them safely.
"""
ensure_parent_dir(target_path)
original_size = os.path.getsize(source_path) if os.path.exists(source_path) else 0
try:
from PIL import Image, ImageOps
with Image.open(source_path) as img:
img.verify()
with Image.open(source_path) as img:
img = ImageOps.exif_transpose(img)
original_dimensions = tuple(img.size)
if img.mode in ("RGBA", "LA") or (img.mode == "P" and "transparency" in img.info):
rgba = img.convert("RGBA")
background = Image.new("RGBA", rgba.size, (255, 255, 255, 255))
background.alpha_composite(rgba)
img = background.convert("RGB")
elif img.mode != "RGB":
img = img.convert("RGB")
resized = False
max_edge = max(img.size) if img.size else 0
if IMAGE_OPTIMIZER_MAX_LONG_EDGE and max_edge > IMAGE_OPTIMIZER_MAX_LONG_EDGE:
scale = IMAGE_OPTIMIZER_MAX_LONG_EDGE / float(max_edge)
new_size = (max(1, int(round(img.size[0] * scale))), max(1, int(round(img.size[1] * scale))))
img = img.resize(new_size, Image.Resampling.LANCZOS)
resized = True
tmp_candidates = []
qualities = IMAGE_OPTIMIZER_QUALITIES or (88, 84, 80)
for quality in qualities:
tmp_candidate = f"{target_path}.q{quality}.tmp"
img.save(tmp_candidate, "JPEG", quality=int(quality), optimize=True, progressive=True)
tmp_candidates.append((quality, tmp_candidate, os.path.getsize(tmp_candidate)))
if os.path.getsize(tmp_candidate) <= IMAGE_OPTIMIZER_TARGET_BYTES:
break
# Keep the highest-quality candidate that reaches target size; otherwise keep the smallest candidate.
under_target = [item for item in tmp_candidates if item[2] <= IMAGE_OPTIMIZER_TARGET_BYTES]
chosen = under_target[0] if under_target else min(tmp_candidates, key=lambda item: item[2])
for quality, path, _size in tmp_candidates:
if path != chosen[1]:
try:
os.remove(path)
except Exception:
pass
os.replace(chosen[1], target_path)
final_size = os.path.getsize(target_path)
return {
"mode": "optimized_jpg",
"quality": chosen[0],
"original_size": original_size,
"final_size": final_size,
"original_dimensions": original_dimensions,
"final_dimensions": tuple(img.size),
"resized": resized,
"saved_bytes": max(0, original_size - final_size),
}
except ImportError:
if is_likely_jpeg_file(source_path):
tmp_target = target_path + ".tmp"
shutil.copy2(source_path, tmp_target)
os.replace(tmp_target, target_path)
final_size = os.path.getsize(target_path)
return {
"mode": "copied_real_jpeg_without_pillow",
"quality": "source",
"original_size": original_size,
"final_size": final_size,
"original_dimensions": "unknown",
"final_dimensions": "unknown",
"resized": False,
"saved_bytes": max(0, original_size - final_size),
}
raise ValueError("Pillow is not installed, so non-JPG images cannot be converted. Install Pillow with: py -m pip install pillow")
except ValueError:
raise
except Exception as exc:
raise ValueError("Image conversion/compression failed: " + repr(exc))
def convert_or_copy_audio_to_mp3(source_path, target_path, original_name=""):
"""Store uploaded audio as a real .mp3 file.
If FFmpeg exists, transcode supported audio/video containers to MP3.
If FFmpeg is missing, accept only files that are already real MP3 bytes, even if their extension is wrong.
"""
ensure_parent_dir(target_path)
original_size = os.path.getsize(source_path) if os.path.exists(source_path) else 0
if original_size <= 0:
raise ValueError("קובץ האודיו ריק.")
if original_size > AUDIO_UPLOAD_MAX_BYTES:
raise ValueError(f"קובץ האודיו גדול מדי: {format_bytes(original_size)}. מגבלה: {format_bytes(AUDIO_UPLOAD_MAX_BYTES)}")
source_is_mp3 = is_likely_mp3_file(source_path)
ffmpeg = find_ffmpeg_executable()
if ffmpeg:
tmp_mp3 = target_path + ".ffmpeg.tmp.mp3"
cmd = [
ffmpeg,
"-y",
"-hide_banner",
"-loglevel", "error",
"-i", source_path,
"-vn",
"-codec:a", "libmp3lame",
"-b:a", AUDIO_MP3_BITRATE,
"-ar", AUDIO_MP3_SAMPLE_RATE,
tmp_mp3,
]
run = run_hidden_subprocess(cmd, timeout=180)
if run.returncode != 0 or not os.path.exists(tmp_mp3) or os.path.getsize(tmp_mp3) <= 0:
try:
if os.path.exists(tmp_mp3):
os.remove(tmp_mp3)
except Exception:
pass
raise ValueError("FFmpeg failed to convert audio to MP3: " + ((run.stderr or run.stdout or "unknown error")[-1200:]))
converted_size = os.path.getsize(tmp_mp3)
if source_is_mp3 and converted_size > int(original_size * 1.05):
# Avoid bloating an already efficient MP3. Still normalize the target filename to .mp3.
try:
os.remove(tmp_mp3)
except Exception:
pass
tmp_target = target_path + ".copy.tmp"
shutil.copy2(source_path, tmp_target)
os.replace(tmp_target, target_path)
final_size = os.path.getsize(target_path)
return {"mode": "copied_existing_mp3_smaller_than_reencode", "original_size": original_size, "final_size": final_size, "saved_bytes": max(0, original_size - final_size), "ffmpeg": ffmpeg}
os.replace(tmp_mp3, target_path)
final_size = os.path.getsize(target_path)
return {"mode": "converted_to_mp3_ffmpeg", "original_size": original_size, "final_size": final_size, "saved_bytes": max(0, original_size - final_size), "ffmpeg": ffmpeg, "bitrate": AUDIO_MP3_BITRATE}
if source_is_mp3:
tmp_target = target_path + ".copy.tmp"
shutil.copy2(source_path, tmp_target)
os.replace(tmp_target, target_path)
final_size = os.path.getsize(target_path)
return {"mode": "copied_real_mp3_without_ffmpeg", "original_size": original_size, "final_size": final_size, "saved_bytes": max(0, original_size - final_size), "ffmpeg": "not_found"}
raise ValueError("FFmpeg is not installed, so this audio file cannot be converted to MP3. Install FFmpeg or upload a real MP3 file.")
def save_audio_upload_to_target(uploaded_file, target_path, backup_group="audio"):
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_name = secure_filename(uploaded_file.filename or "audio_upload") or "audio_upload"
tmp_path = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"audio_{stamp}_{os.getpid()}_{safe_name}")
uploaded_file.save(tmp_path)
try:
backup_path = backup_existing_file(target_path, backup_group)
media = convert_or_copy_audio_to_mp3(tmp_path, target_path, uploaded_file.filename or "")
return {"path": target_path, "size": os.path.getsize(target_path), "sha256": file_sha256(target_path), "backup": backup_path, "media": media}
finally:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
def save_audio_upload_to_multiple_targets(uploaded_file, base_folder, rel_paths, backup_group):
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_name = secure_filename(uploaded_file.filename or "audio_upload") or "audio_upload"
source_tmp = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"audio_multi_{stamp}_{os.getpid()}_{safe_name}")
converted_tmp = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"audio_multi_{stamp}_{os.getpid()}_converted.mp3")
uploaded_file.save(source_tmp)
try:
media = convert_or_copy_audio_to_mp3(source_tmp, converted_tmp, uploaded_file.filename or "")
sha = file_sha256(converted_tmp)
results = []
for rel_path in rel_paths:
target_path = os.path.join(base_folder, rel_path)
ensure_parent_dir(target_path)
backup_path = backup_existing_file(target_path, backup_group)
tmp_target = target_path + ".tmp"
shutil.copy2(converted_tmp, tmp_target)
os.replace(tmp_target, target_path)
results.append({"rel": rel_path, "path": target_path, "size": os.path.getsize(target_path), "sha256": file_sha256(target_path), "backup": backup_path})
return {"size": os.path.getsize(converted_tmp), "sha256": sha, "targets": results, "media": media}
finally:
for p in (source_tmp, converted_tmp):
try:
if os.path.exists(p):
os.remove(p)
except Exception:
pass
def cleanup_old_pending_music_uploads():
try:
os.makedirs(PENDING_MUSIC_UPLOAD_DIR, exist_ok=True)
now = time.time()
for name in os.listdir(PENDING_MUSIC_UPLOAD_DIR):
path = os.path.join(PENDING_MUSIC_UPLOAD_DIR, name)
try:
if os.path.isfile(path) and now - os.path.getmtime(path) > PENDING_MUSIC_MAX_AGE_SECONDS:
os.remove(path)
except Exception:
pass
except Exception:
pass
def save_pending_audio_upload(uploaded_file, trip, music_key):
cleanup_old_pending_music_uploads()
os.makedirs(PENDING_MUSIC_UPLOAD_DIR, exist_ok=True)
token = uuid.uuid4().hex
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_name = secure_filename(uploaded_file.filename or "audio_upload") or "audio_upload"
source_tmp = os.path.join(PENDING_MUSIC_UPLOAD_DIR, f"{token}_{stamp}_{safe_name}")
pending_mp3 = os.path.join(PENDING_MUSIC_UPLOAD_DIR, f"{token}.mp3")
meta_path = os.path.join(PENDING_MUSIC_UPLOAD_DIR, f"{token}.json")
uploaded_file.save(source_tmp)
try:
media = convert_or_copy_audio_to_mp3(source_tmp, pending_mp3, uploaded_file.filename or "")
meta = {
"token": token,
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"created_epoch": time.time(),
"trip": (trip or "").upper(),
"music_key": music_key,
"original_name": uploaded_file.filename or "",
"pending_path": pending_mp3,
"size": os.path.getsize(pending_mp3),
"sha256": file_sha256(pending_mp3),
"media": media,
}
with open(meta_path, "w", encoding="utf-8") as f:
json.dump(meta, f, ensure_ascii=False, indent=2)
return meta
except Exception:
for p in (pending_mp3, meta_path):
try:
if os.path.exists(p):
os.remove(p)
except Exception:
pass
raise
finally:
try:
if os.path.exists(source_tmp):
os.remove(source_tmp)
except Exception:
pass
def read_pending_music_upload(token):
token = (token or "").strip()
if not re.fullmatch(r"[0-9a-fA-F]{32}", token):
raise ValueError("אסימון העלאה זמנית לא תקין.")
meta_path = os.path.join(PENDING_MUSIC_UPLOAD_DIR, f"{token}.json")
if not os.path.exists(meta_path):
raise FileNotFoundError("העלאת השיר הזמנית לא נמצאה. ייתכן שפג תוקפה או שהיא כבר טופלה.")
with open(meta_path, "r", encoding="utf-8") as f:
meta = json.load(f)
if not isinstance(meta, dict) or meta.get("token") != token:
raise ValueError("קובץ ההעלאה הזמנית אינו תקין.")
pending_path = meta.get("pending_path") or os.path.join(PENDING_MUSIC_UPLOAD_DIR, f"{token}.mp3")
if not os.path.exists(pending_path):
raise FileNotFoundError("קובץ השיר הזמני לא נמצא.")
if time.time() - float(meta.get("created_epoch") or 0) > PENDING_MUSIC_MAX_AGE_SECONDS:
discard_pending_music_upload(token)
raise FileNotFoundError("פג תוקף העלאת השיר הזמנית. העלה את השיר מחדש.")
meta["pending_path"] = pending_path
return meta
def discard_pending_music_upload(token):
token = (token or "").strip()
if not re.fullmatch(r"[0-9a-fA-F]{32}", token):
return
for suffix in (".mp3", ".json"):
path = os.path.join(PENDING_MUSIC_UPLOAD_DIR, token + suffix)
try:
if os.path.exists(path):
os.remove(path)
except Exception:
pass
def commit_pending_audio_to_multiple_targets(pending_meta, base_folder, rel_paths, backup_group):
pending_path = pending_meta.get("pending_path")
if not pending_path or not os.path.exists(pending_path):
raise FileNotFoundError("קובץ השיר הזמני לא נמצא ולכן לא ניתן להטמיע אותו.")
sha = file_sha256(pending_path)
size = os.path.getsize(pending_path)
results = []
for rel_path in rel_paths:
target_path = os.path.join(base_folder, rel_path)
ensure_parent_dir(target_path)
backup_path = backup_existing_file(target_path, backup_group)
tmp_target = target_path + ".tmp"
shutil.copy2(pending_path, tmp_target)
os.replace(tmp_target, target_path)
results.append({"rel": rel_path, "path": target_path, "size": os.path.getsize(target_path), "sha256": file_sha256(target_path), "backup": backup_path})
return {"size": size, "sha256": sha, "targets": results, "media": pending_meta.get("media"), "pending_token": pending_meta.get("token")}
def build_japan_music_duration_status(music_key, save_result, probe, sync_result=None, sync_note="", sync_error=None, approval_action=""):
rule_key = music_key_to_rule_key("JAPAN", music_key)
rule = PRESENTATION_MUSIC_SYNC_RULES.get(rule_key or "", {})
status = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"trip": "JAPAN",
"music_key": music_key,
"rule_key": rule_key or "",
"label": rule.get("label", ""),
"scope": rule.get("scope", ""),
"presentation": rule.get("presentation", ""),
"paths": all_saved_audio_paths(save_result),
"primary_path": primary_saved_audio_path(save_result) or "",
"duration_ok": bool(probe and probe.get("ok")),
"duration_seconds": (probe or {}).get("duration_seconds"),
"duration_display": (probe or {}).get("duration_display", ""),
"ffprobe": (probe or {}).get("ffprobe", ""),
"sync_performed": bool(sync_result),
"sync_note": sync_note,
"approval_action": approval_action,
}
if sync_result:
status["sync_results"] = [sync_result]
status["sync_summary"] = sync_result.get("message", "") if isinstance(sync_result, dict) else ""
if sync_error:
status["sync_error"] = repr(sync_error)
if rule_key:
write_presentation_music_upload_status(rule_key, status)
return status
def japan_experience_music_analysis(duration_seconds):
data = get_japan_experience_slide_durations_ms()
slide_count = len(data.get("durations") or [])
if slide_count <= 0:
raise ValueError("לא נמצאו שקפים במצגת יפן.")
duration = float(duration_seconds)
average = duration / float(slide_count)
min_total = slide_count * JAPAN_EXPERIENCE_MIN_AVG_SLIDE_SECONDS
max_total = slide_count * JAPAN_EXPERIENCE_MAX_AVG_SLIDE_SECONDS
if average < JAPAN_EXPERIENCE_MIN_AVG_SLIDE_SECONDS:
state = "too_short"
elif average > JAPAN_EXPERIENCE_MAX_AVG_SLIDE_SECONDS:
state = "too_long"
else:
state = "ok"
return {
"state": state,
"slide_count": slide_count,
"song_seconds": duration,
"song_display": format_seconds(duration),
"average_slide_seconds": average,
"minimum_total_seconds": min_total,
"maximum_total_seconds": max_total,
"loop_total_seconds": min_total,
"presentation_path": data.get("path", ""),
}
def render_japan_music_confirmation_page(token, analysis, probe, pending_meta):
state = analysis.get("state")
avg = float(analysis.get("average_slide_seconds") or 0)
slide_count = int(analysis.get("slide_count") or 0)
song_display = html_lib.escape(str(analysis.get("song_display") or ""))
original_name = html_lib.escape(str(pending_meta.get("original_name") or ""))
loop_total = html_lib.escape(format_seconds(analysis.get("loop_total_seconds") or 0))
token_safe = html_lib.escape(token)
if state == "too_short":
title = "נדרש אישור לפני הטמעת השיר"
badge = "אזהרה: השיר קצר מדי ביחס למספר השקפים"
message = (
f"השיר שזוהה באורך {song_display}. במצגת יש {slide_count} שקפים, "
f"ולכן זמן ממוצע לשקף יהיה {avg:.2f} שניות — פחות מהמינימום שהוגדר: "
f"{JAPAN_EXPERIENCE_MIN_AVG_SLIDE_SECONDS:.0f} שניות לשקף."
)
options = f"""
<form action=\"/confirm_japan_experience_music_upload/{token_safe}\" method=\"post\"><input type=\"hidden\" name=\"action\" value=\"loop_min_5\"><button class=\"primary\">הטמע את השיר בלופ ושמור לפחות 5 שניות לכל שקף</button></form>
<form action=\"/confirm_japan_experience_music_upload/{token_safe}\" method=\"post\"><input type=\"hidden\" name=\"action\" value=\"accept_as_is\"><button class=\"warn\">הטמע בכל זאת וסנכרן לפי אורך השיר המקורי</button></form>
<form action=\"/confirm_japan_experience_music_upload/{token_safe}\" method=\"post\"><input type=\"hidden\" name=\"action\" value=\"cancel\"><button class=\"danger\">אל תטמיע את השיר הזה</button></form>
"""
note = f"בחירת הלופ תאריך את המצגת ל־{loop_total} לפחות. קובץ האודיו עצמו יישמר, ותגית ה־audio במצגת תמשיך לנגן אותו בלופ."
elif state == "too_long":
title = "נדרש אישור לפני הטמעת השיר"
badge = "אזהרה: השיר ארוך מדי ביחס למספר השקפים"
message = (
f"השיר שזוהה באורך {song_display}. במצגת יש {slide_count} שקפים, "
f"ולכן זמן ממוצע לשקף יהיה {avg:.2f} שניות — מעל המקסימום שהוגדר: "
f"{JAPAN_EXPERIENCE_MAX_AVG_SLIDE_SECONDS:.0f} שניות לשקף."
)
options = f"""
<form action=\"/confirm_japan_experience_music_upload/{token_safe}\" method=\"post\"><input type=\"hidden\" name=\"action\" value=\"accept_as_is\"><button class=\"warn\">הטמע בכל זאת וסנכרן לפי אורך השיר</button></form>
<form action=\"/confirm_japan_experience_music_upload/{token_safe}\" method=\"post\"><input type=\"hidden\" name=\"action\" value=\"cancel\"><button class=\"danger\">אל תטמיע את השיר הזה</button></form>
"""
note = "במצב הזה אין אפשרות לופ, כי הבעיה אינה שיר קצר אלא שיר ארוך מדי ביחס למצגת."
else:
return admin_status_page("אין צורך באישור", "משך השיר נמצא בטווח התקין. העלה שוב אם נדרש.", ok=True)
return f"""
<!DOCTYPE html>
<html lang=\"he\" dir=\"rtl\"><head><meta charset=\"UTF-8\"><meta name=\"viewport\" content=\"width=device-width, initial-scale=1.0\"><title>{html_lib.escape(title)}</title>
<style>
body{{margin:0;font-family:Arial,'Segoe UI',sans-serif;background:linear-gradient(180deg,#0f172a,#020617);color:white;min-height:100vh;display:flex;align-items:center;justify-content:center;padding:20px}}
.card{{width:min(820px,94vw);background:#111827;border:1px solid #334155;border-radius:28px;padding:26px;box-shadow:0 24px 90px rgba(0,0,0,.45)}}
.badge{{display:inline-flex;background:#7c2d12;color:#fed7aa;border:1px solid #fb923c;border-radius:999px;padding:8px 14px;font-weight:900;margin-bottom:12px}}
h1{{margin:0 0 12px;font-size:clamp(28px,5vw,42px)}}
p{{color:#e5e7eb;line-height:1.7;font-size:17px}}
.meta{{background:#020617;border:1px solid #334155;border-radius:18px;padding:14px;margin:14px 0;color:#bfdbfe;line-height:1.7}}
.actions{{display:grid;gap:10px;margin-top:16px}}
button,a{{border:0;border-radius:14px;padding:14px 16px;color:white;font-weight:950;cursor:pointer;text-decoration:none;font-size:16px;width:100%}}
.primary{{background:#16a34a}}.warn{{background:#ca8a04}}.danger{{background:#dc2626}}.back{{display:inline-flex;width:auto;background:#334155;margin-top:12px}}
form{{margin:0}}
.usa-image-target-panel{display:none;margin-top:16px}.usa-image-target-panel.active{display:block}.image-target-choice .sub-card{border-color:rgba(96,165,250,.28)}.tool-head.compact{margin-top:10px;padding-top:4px}.tool-head.compact h3{margin-top:0}code{direction:ltr;display:inline-block;background:rgba(15,23,42,.65);border:1px solid rgba(148,163,184,.25);border-radius:8px;padding:2px 6px;color:#dbeafe}
</style></head><body><div class=\"card\"><span class=\"badge\">{html_lib.escape(badge)}</span><h1>{html_lib.escape(title)}</h1><p>{html_lib.escape(message)}</p><div class=\"meta\"><b>קובץ:</b> {original_name or '—'}<br><b>משך שיר:</b> {song_display}<br><b>שקפים:</b> {slide_count}<br><b>ממוצע צפוי לשקף:</b> {avg:.2f} שניות<br><b>סטטוס:</b> השיר עדיין לא הוטמע בנתיבי המצגת. הוא נשמר זמנית בלבד עד לבחירה.</div><p>{html_lib.escape(note)}</p><div class=\"actions\">{options}</div><a class=\"back\" href=\"/admin#japan\">חזרה בלי לבצע שינוי</a></div></body></html>
"""
def handle_japan_experience_music_upload_with_confirmation(uploaded_file):
pending_meta = None
try:
pending_meta = save_pending_audio_upload(uploaded_file, "JAPAN", "japan_experience_music")
probe = probe_audio_duration_seconds(pending_meta.get("pending_path"))
if not probe.get("ok"):
discard_pending_music_upload(pending_meta.get("token"))
details = media_details_text(pending_meta.get("media"))
details += "\n\nשגיאת ffprobe:\n" + str(probe.get("error") or probe.get("raw") or "לא ידוע")
return admin_status_page("השיר לא הוטמע — בדיקת אורך נכשלה", "המערכת לא הצליחה לזהות משך תקין ולכן לא שמרה את השיר בנתיבי המצגת ולא שינתה את המצגת.", ok=False, details=details), 200
analysis = japan_experience_music_analysis(probe.get("duration_seconds"))
if analysis.get("state") in {"too_short", "too_long"}:
return render_japan_music_confirmation_page(pending_meta.get("token"), analysis, probe, pending_meta), 200
return commit_japan_experience_pending_music(pending_meta.get("token"), "accept_as_is")
except Exception as exc:
if pending_meta and pending_meta.get("token"):
discard_pending_music_upload(pending_meta.get("token"))
return admin_status_page("שגיאה בהכנת שיר יפן", "השיר לא הוטמע ולא בוצע שינוי במצגת.", ok=False, details=repr(exc)), 500
def commit_japan_experience_pending_music(token, action):
try:
action = (action or "").strip()
pending_meta = read_pending_music_upload(token)
if pending_meta.get("trip") != "JAPAN" or pending_meta.get("music_key") != "japan_experience_music":
return admin_status_page("העלאה זמנית לא תואמת", "הקובץ הזמני אינו שייך לשיר מצגת יפן הראשית.", ok=False), 400
probe = probe_audio_duration_seconds(pending_meta.get("pending_path"))
if not probe.get("ok"):
discard_pending_music_upload(token)
return admin_status_page("השיר לא הוטמע — בדיקת אורך נכשלה", "הקובץ הזמני קיים, אבל ffprobe לא הצליח לזהות את משך השיר בזמן האישור. לא בוצע שינוי.", ok=False, details=str(probe.get("error") or probe.get("raw") or "לא ידוע")), 200
analysis = japan_experience_music_analysis(probe.get("duration_seconds"))
allowed = {"accept_as_is", "cancel"}
if analysis.get("state") == "too_short":
allowed.add("loop_min_5")
if action not in allowed:
return admin_status_page("בחירת סנכרון לא תקינה", "האפשרות שנשלחה אינה תואמת למצב השיר.", ok=False), 400
if action == "cancel":
discard_pending_music_upload(token)
status = build_japan_music_duration_status("japan_experience_music", {"targets": []}, probe, sync_result=None, sync_note="המשתמש ביטל את ההעלאה אחרי אזהרת זמן שקפים. השיר לא הוטמע והמצגת לא שונתה.", approval_action="cancel")
return admin_status_page("העלאת השיר בוטלה", "השיר לא הוטמע בנתיבי המצגת והמצגת לא שונתה.", ok=True, details=music_upload_duration_details(status))
save_result = commit_pending_audio_to_multiple_targets(pending_meta, FOLDERS["JAPAN"], JAPAN_MUSIC_TARGETS["japan_experience_music"], "JAPAN_music")
if action == "loop_min_5":
target_seconds = max(float(analysis.get("loop_total_seconds") or 0), float(probe.get("duration_seconds") or 0))
sync_result = sync_japan_experience_presentation_to_target_duration(
target_seconds,
sync_mode="short_song_audio_loop_min_5_seconds_per_slide",
source_duration_seconds=probe.get("duration_seconds"),
minimum_per_slide_seconds=JAPAN_EXPERIENCE_LOOP_MIN_SLIDE_SECONDS,
)
sync_note = "השיר הוטמע, אך לא דחס את המצגת. המצגת הוארכה כך שלכל שקף יש לפחות 5 שניות, והשיר יתנגן בלופ."
title = "השיר הוטמע בלופ והמצגת סונכרנה בבטחה"
message = "השיר נשמר בנתיבי המצגת. בגלל שהוא קצר מדי, זמני השקפים נקבעו לרצפת 5 שניות לשקף והמוזיקה תמשיך בלופ."
else:
sync_result = sync_japan_experience_presentation_to_music_duration(probe.get("duration_seconds"))
if analysis.get("state") == "too_short":
sync_note = "השיר הוטמע לאחר אישור מפורש למרות שהוא יוצר פחות מ־5 שניות בממוצע לשקף."
title = "השיר הוטמע לפי האורך המקורי"
message = "השיר נשמר והמצגת סונכרנה לפי אורך השיר המקורי, בהתאם לאישור שניתן במסך האזהרה."
elif analysis.get("state") == "too_long":
sync_note = "השיר הוטמע לאחר אישור מפורש למרות שהוא יוצר מעל 10 שניות בממוצע לשקף."
title = "השיר הארוך הוטמע לפי האורך המקורי"
message = "השיר נשמר והמצגת סונכרנה לפי אורך השיר הארוך, בהתאם לאישור שניתן במסך האזהרה."
else:
sync_note = "השיר נמצא בטווח התקין 5–10 שניות בממוצע לשקף, ולכן הוטמע וסונכרן ללא אזהרה."
title = "מוזיקת יפן הועלתה והמצגת סונכרנה"
message = "השיר נשמר, משך השיר זוהה, וזמני השקפים עודכנו לפי האורך החדש."
status = build_japan_music_duration_status("japan_experience_music", save_result, probe, sync_result=sync_result, sync_note=sync_note, approval_action=action)
discard_pending_music_upload(token)
details = "נשמר אל:\n" + "\n".join([item["path"] for item in save_result["targets"]]) + f"\n\nSHA256 MP3 סופי:\n{save_result['sha256']}"
details += "\n\n" + media_details_text(save_result.get("media"))
details += "\n\n" + music_upload_duration_details(status)
return admin_status_page(title, message, ok=True, details=details)
except Exception as exc:
try:
discard_pending_music_upload(token)
except Exception:
pass
return admin_status_page("שגיאה באישור הטמעת שיר", "השיר לא הוטמע או שהפעולה לא הושלמה במלואה. בדוק את הפירוט לפני ניסיון חוזר.", ok=False, details=repr(exc)), 500
def media_details_text(media):
if not media:
return ""
lines = []
if media.get("mode"):
lines.append(f"מצב עיבוד: {media.get('mode')}")
if media.get("original_size") is not None and media.get("final_size") is not None:
lines.append(f"גודל לפני: {format_bytes(media.get('original_size'))}")
lines.append(f"גודל אחרי: {format_bytes(media.get('final_size'))}")
saved = int(media.get("saved_bytes") or 0)
if saved > 0:
lines.append(f"חיסכון: {format_bytes(saved)}")
if media.get("original_dimensions"):
lines.append(f"מידות מקור: {media.get('original_dimensions')}")
if media.get("final_dimensions"):
lines.append(f"מידות סופיות: {media.get('final_dimensions')}")
if media.get("quality"):
lines.append(f"איכות JPG: {media.get('quality')}")
if media.get("bitrate"):
lines.append(f"ביטרייט MP3: {media.get('bitrate')}")
if media.get("duration_seconds") is not None:
lines.append(f"משך סרטון: {media.get('duration_display') or format_seconds(media.get('duration_seconds'))} ({float(media.get('duration_seconds') or 0):.3f} שניות)")
if media.get("codec"):
lines.append(f"קידוד וידאו: {media.get('codec')}")
if media.get("width") and media.get("height"):
lines.append(f"רזולוציית וידאו: {media.get('width')}×{media.get('height')}")
if media.get("ffmpeg"):
lines.append(f"FFmpeg: {media.get('ffmpeg')}")
return "\n".join(lines)
# ================= USA VIDEO MEDIA ENGINE v34.33 =================
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
VIDEO_EXTENSIONS = {".mp4", ".m4v"}
MEDIA_EXTENSIONS = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS
MANIFEST_NAME = USA_MEDIA_MANIFEST_NAME
PATCH_VERSION = USA_MEDIA_PATCH_VERSION
MIN_IMAGE_SECONDS = 0.05
EPSILON = 0.035
class MediaError(RuntimeError):
"""Raised when media or presentation validation fails."""
@dataclass(frozen=True)
class PresentationPart:
start: int
end: int
seconds: float
label: str
@dataclass(frozen=True)
class PresentationInfo:
slide_bases: Tuple[str, ...]
parts: Tuple[PresentationPart, ...]
part1_last_slide_number: int
@dataclass(frozen=True)
class TimelineEntry:
index: int
base: str
kind: str
media_seconds: float
timeline_seconds: float
start: float
end: float
part_index: int
@dataclass(frozen=True)
class TimelineResult:
entries: Tuple[TimelineEntry, ...]
total_seconds: float
warnings: Tuple[str, ...]
# --------------------------- MP4 duration parser ---------------------------
def _read_u32(f) -> int:
data = f.read(4)
if len(data) != 4:
raise MediaError("קובץ הווידאו נקטע בזמן קריאת מבנה MP4.")
return struct.unpack(">I", data)[0]
def _read_u64(f) -> int:
data = f.read(8)
if len(data) != 8:
raise MediaError("קובץ הווידאו נקטע בזמן קריאת מבנה MP4.")
return struct.unpack(">Q", data)[0]
def _iter_boxes(f, start: int, end: int):
pos = start
while pos + 8 <= end:
f.seek(pos)
size = _read_u32(f)
box_type = f.read(4)
header = 8
if size == 1:
size = _read_u64(f)
header = 16
elif size == 0:
size = end - pos
if size < header or pos + size > end:
break
yield box_type, pos + header, pos + size
pos += size
def mp4_duration_seconds(path: Path) -> float:
"""Read duration from the ISO BMFF mvhd atom without ffmpeg."""
path = Path(path)
if not path.is_file():
raise MediaError(f"קובץ וידאו לא נמצא: {path.name}")
file_size = path.stat().st_size
if file_size < 24:
raise MediaError(f"קובץ וידאו קטן/פגום: {path.name}")
with path.open("rb") as f:
moov = None
for box_type, content_start, box_end in _iter_boxes(f, 0, file_size):
if box_type == b"moov":
moov = (content_start, box_end)
break
if not moov:
raise MediaError(f"לא נמצא אטום moov בקובץ {path.name}. יש לשמור MP4 תקין עם faststart.")
for box_type, content_start, box_end in _iter_boxes(f, moov[0], moov[1]):
if box_type != b"mvhd":
continue
f.seek(content_start)
version_flags = f.read(4)
if len(version_flags) != 4:
break
version = version_flags[0]
if version == 1:
f.seek(content_start + 4 + 8 + 8)
timescale = _read_u32(f)
duration = _read_u64(f)
else:
f.seek(content_start + 4 + 4 + 4)
timescale = _read_u32(f)
duration = _read_u32(f)
if timescale <= 0 or duration <= 0:
raise MediaError(f"משך וידאו לא תקין בקובץ {path.name}.")
seconds = duration / timescale
if not math.isfinite(seconds) or seconds <= 0 or seconds > 24 * 60 * 60:
raise MediaError(f"משך וידאו לא סביר בקובץ {path.name}: {seconds}")
return float(seconds)
raise MediaError(f"לא נמצא משך וידאו תקין בקובץ {path.name}.")
# ------------------------- Presentation inspection ------------------------
_SLIDE_SECTION_RE = re.compile(
r"<section\b(?=[^>]*\bclass=[\"'][^\"']*\bslide\b[^\"']*[\"'])[^>]*>(.*?)</section>",
re.IGNORECASE | re.DOTALL,
)
_DATA_LOCAL_BASE_RE = re.compile(r"\bdata-local-base=[\"']([^\"']+)[\"']", re.IGNORECASE)
_IMG_SRC_RE = re.compile(r"<img\b[^>]*\bsrc=[\"']([^\"']+)[\"']", re.IGNORECASE | re.DOTALL)
def _base_from_slide_html(section_html: str, index: int) -> str:
match = _DATA_LOCAL_BASE_RE.search(section_html)
if match:
return Path(match.group(1).replace("\\", "/")).name
src = _IMG_SRC_RE.search(section_html)
if src:
return Path(src.group(1).split("?", 1)[0].replace("\\", "/")).stem
return f"slide-{index + 1:02d}"
def inspect_presentation_html(html: str) -> PresentationInfo:
slide_sections = list(_SLIDE_SECTION_RE.finditer(html))
slide_bases = tuple(_base_from_slide_html(m.group(1), i) for i, m in enumerate(slide_sections))
if not slide_bases:
# Fallback for malformed/minified HTML: count class=slide openings.
starts = list(re.finditer(r"<(?:section|div)\b[^>]*\bclass=[\"'][^\"']*\bslide\b", html, re.I))
slide_bases = tuple(f"slide-{i + 1:02d}" for i in range(len(starts)))
if not slide_bases:
raise MediaError("לא נמצאו שקפים בקובץ המצגת.")
def number(name: str) -> float:
m = re.search(rf"\bconst\s+{re.escape(name)}\s*=\s*([0-9]+(?:\.[0-9]+)?)\s*;", html)
if not m:
raise MediaError(f"לא נמצא המשתנה {name} בקובץ המצגת.")
return float(m.group(1))
part1_seconds = number("PART_1_SECONDS")
part2_seconds = number("PART_2_SECONDS")
part1_last = int(round(number("PART_1_LAST_SLIDE_NUMBER")))
if not 1 <= part1_last < len(slide_bases):
raise MediaError(
f"חלוקת השירים אינה תקינה: PART_1_LAST_SLIDE_NUMBER={part1_last}, מספר שקפים={len(slide_bases)}."
)
parts = (
PresentationPart(0, part1_last - 1, part1_seconds, "שיר ראשון"),
PresentationPart(part1_last, len(slide_bases) - 1, part2_seconds, "שיר שני"),
)
return PresentationInfo(slide_bases=slide_bases, parts=parts, part1_last_slide_number=part1_last)
def inspect_presentation_file(path: Path) -> PresentationInfo:
return inspect_presentation_html(Path(path).read_text(encoding="utf-8"))
# ---------------------------- Timeline allocator ---------------------------
def allocate_timeline(
slide_bases: Sequence[str],
parts: Sequence[PresentationPart],
video_durations: Mapping[str, float],
*,
min_image_seconds: float = MIN_IMAGE_SECONDS,
) -> TimelineResult:
if not slide_bases:
raise MediaError("אין שקפים לחישוב ציר זמן.")
normalized_videos = {str(k).lower(): float(v) for k, v in video_durations.items()}
durations = [0.0] * len(slide_bases)
media_seconds = [0.0] * len(slide_bases)
kinds = ["image"] * len(slide_bases)
warnings: List[str] = []
for part_index, part in enumerate(parts):
indices = list(range(part.start, part.end + 1))
if not indices:
continue
video_indices = []
for i in indices:
duration = normalized_videos.get(slide_bases[i].lower())
if duration is not None:
if not math.isfinite(duration) or duration <= 0:
raise MediaError(f"משך וידאו לא תקין בשקף {i + 1}: {duration}")
video_indices.append(i)
media_seconds[i] = duration
kinds[i] = "video"
image_indices = [i for i in indices if i not in video_indices]
video_total = sum(media_seconds[i] for i in video_indices)
if not video_indices:
equal = part.seconds / len(indices)
for i in indices:
durations[i] = equal
continue
if video_total > part.seconds + EPSILON:
raise MediaError(
f"{part.label}: אורך הסרטונים המצטבר {video_total:.3f} שנ׳ ארוך ממשך השיר "
f"{part.seconds:.3f} שנ׳. אי אפשר להציג את כולם עד הסוף בלי לחרוג מהמוזיקה."
)
remaining = max(0.0, part.seconds - video_total)
if image_indices:
per_image = remaining / len(image_indices)
if per_image < min_image_seconds:
raise MediaError(
f"{part.label}: לאחר ניכוי הסרטונים נשארו רק {remaining:.3f} שנ׳ "
f"ל־{len(image_indices)} שקפי תמונה ({per_image:.3f} שנ׳ לשקף)."
)
for i in video_indices:
durations[i] = media_seconds[i]
for i in image_indices:
durations[i] = per_image
if per_image < 1.0:
warnings.append(
f"{part.label}: שקפי התמונות יוצגו {per_image:.2f} שנ׳ בלבד בגלל כמות/אורך הסרטונים."
)
else:
# All slides are videos. Preserve complete playback and spread any spare song time
# as a last-frame hold. This keeps the total exactly equal to the song.
hold = remaining / len(video_indices)
for i in video_indices:
durations[i] = media_seconds[i] + hold
if hold > EPSILON:
warnings.append(
f"{part.label}: כל השקפים הם סרטונים; לאחר כל סרטון תישאר תמונת הסיום עוד {hold:.2f} שנ׳."
)
# Correct floating-point residue on the final slide of the part.
residue = part.seconds - sum(durations[i] for i in indices)
durations[indices[-1]] += residue
entries: List[TimelineEntry] = []
cursor = 0.0
part_for_index = {}
for p_i, part in enumerate(parts):
for i in range(part.start, part.end + 1):
part_for_index[i] = p_i
for i, base in enumerate(slide_bases):
start = cursor
cursor += durations[i]
entries.append(
TimelineEntry(
index=i,
base=base,
kind=kinds[i],
media_seconds=media_seconds[i],
timeline_seconds=durations[i],
start=start,
end=cursor,
part_index=part_for_index.get(i, 0),
)
)
expected = sum(p.seconds for p in parts)
if abs(cursor - expected) > 0.005:
raise MediaError(f"כשל פנימי בחישוב ציר הזמן: {cursor:.6f} במקום {expected:.6f} שנ׳.")
return TimelineResult(entries=tuple(entries), total_seconds=expected, warnings=tuple(warnings))
# ------------------------------ Media manifest -----------------------------
def media_files_by_base(media_dir: Path) -> Dict[str, Path]:
media_dir = Path(media_dir)
result: Dict[str, Path] = {}
if not media_dir.is_dir():
return result
# Videos win only when an image sibling was not removed by an older admin version.
candidates = sorted((p for p in media_dir.iterdir() if p.is_file() and p.suffix.lower() in MEDIA_EXTENSIONS), key=lambda p: p.name.lower())
for path in candidates:
key = path.stem.lower()
previous = result.get(key)
if previous is None or path.suffix.lower() in VIDEO_EXTENSIONS:
result[key] = path
return result
def build_manifest(media_dir: Path, slide_bases: Optional[Sequence[str]] = None) -> dict:
files = media_files_by_base(media_dir)
allowed = {b.lower() for b in slide_bases} if slide_bases else None
items = []
for key, path in sorted(files.items()):
if allowed is not None and key not in allowed:
continue
ext = path.suffix.lower()
item = {
"base": path.stem,
"file": path.name,
"type": "video" if ext in VIDEO_EXTENSIONS else "image",
}
if ext in VIDEO_EXTENSIONS:
item["duration"] = round(mp4_duration_seconds(path), 6)
items.append(item)
return {
"version": 2,
"patchVersion": PATCH_VERSION,
"generatedAt": datetime.now().astimezone().isoformat(timespec="seconds"),
"items": items,
}
def write_manifest(media_dir: Path, slide_bases: Optional[Sequence[str]] = None) -> Path:
media_dir = Path(media_dir)
media_dir.mkdir(parents=True, exist_ok=True)
manifest = build_manifest(media_dir, slide_bases)
target = media_dir / MANIFEST_NAME
tmp = target.with_suffix(".json.tmp")
tmp.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(tmp, target)
return target
def video_duration_map(media_dir: Path, slide_bases: Optional[Sequence[str]] = None) -> Dict[str, float]:
manifest = build_manifest(media_dir, slide_bases)
return {
item["base"]: float(item["duration"])
for item in manifest["items"]
if item["type"] == "video"
}
# --------------------------- Presentation patching -------------------------
CSS_PATCH = r"""
/* EITAN_VIDEO_MEDIA_CSS_START — generated by admin */
.hero-video{
width:100%;height:100%;object-fit:cover;display:block;
filter:saturate(1.12) contrast(1.08) brightness(.96);
background:#020617;
}
.hero-video::-webkit-media-controls{display:none!important}
/* EITAN_VIDEO_MEDIA_CSS_END */
""".strip()
JS_RUNTIME_PATCH = r"""
/* EITAN_VIDEO_MEDIA_RUNTIME_START — generated by admin */
const EITAN_VIDEO_MEDIA_PATCH_VERSION = '2026-07-12-video-sync-v1';
const EITAN_MEDIA_MANIFEST_NAME = 'media-manifest.json';
const eitanManifestUrls = new Set();
const eitanVideoDurations = new Map();
let eitanSlideStarts = new Array(slides.length).fill(0);
let eitanSlideEnds = new Array(slides.length).fill(0);
let eitanSlideTimelineDurations = new Array(slides.length).fill(0);
let eitanMediaPrepared = false;
let eitanMediaPreparePromise = null;
function eitanBaseName(value) {
const clean = String(value || '').split('?')[0].replace(/\\/g, '/');
const name = clean.slice(clean.lastIndexOf('/') + 1);
return name.replace(/\.[^.]+$/, '');
}
function eitanBaseForSlide(slide, slideIndex) {
const img = slide && slide.querySelector ? slide.querySelector('img.hero-image') : null;
if (img) return eitanBaseName(img.dataset.localBase || img.getAttribute('src') || '');
const media = slide && slide.querySelector ? slide.querySelector('video.hero-video') : null;
if (media) return String(media.dataset.mediaBase || eitanBaseName(media.dataset.mediaSource || media.src));
return `slide-${slideIndex + 1}`;
}
function eitanBuildTimeline(videoDurationsByIndex = new Map()) {
const durations = new Array(slides.length).fill(0);
const starts = new Array(slides.length).fill(0);
const ends = new Array(slides.length).fill(0);
const tiny = 0.05;
parts.forEach((part, partIndex) => {
const indices = [];
for (let i = part.start; i <= part.end; i++) indices.push(i);
const videoIndices = indices.filter(i => videoDurationsByIndex.has(i));
const imageIndices = indices.filter(i => !videoDurationsByIndex.has(i));
const videoTotal = videoIndices.reduce((sum, i) => sum + Number(videoDurationsByIndex.get(i) || 0), 0);
if (!videoIndices.length) {
const equal = part.seconds / Math.max(1, indices.length);
indices.forEach(i => { durations[i] = equal; });
} else {
if (videoTotal > part.seconds + 0.035) {
throw new Error(`${part.label}: משך הסרטונים (${videoTotal.toFixed(2)} שנ׳) ארוך ממשך השיר (${part.seconds.toFixed(2)} שנ׳).`);
}
const remaining = Math.max(0, part.seconds - videoTotal);
if (imageIndices.length) {
const perImage = remaining / imageIndices.length;
if (perImage < tiny) {
throw new Error(`${part.label}: לא נותר זמן שימושי לשקפי התמונות לאחר ניכוי הסרטונים.`);
}
videoIndices.forEach(i => { durations[i] = Number(videoDurationsByIndex.get(i)); });
imageIndices.forEach(i => { durations[i] = perImage; });
} else {
const hold = remaining / Math.max(1, videoIndices.length);
videoIndices.forEach(i => { durations[i] = Number(videoDurationsByIndex.get(i)) + hold; });
}
const residue = part.seconds - indices.reduce((sum, i) => sum + durations[i], 0);
durations[indices[indices.length - 1]] += residue;
}
});
let cursor = 0;
durations.forEach((duration, i) => {
starts[i] = cursor;
cursor += duration;
ends[i] = cursor;
});
if (Math.abs(cursor - totalDuration) > 0.01) {
throw new Error(`כשל בחישוב הסנכרון: ${cursor.toFixed(3)} במקום ${totalDuration.toFixed(3)} שנ׳.`);
}
eitanSlideTimelineDurations = durations;
eitanSlideStarts = starts;
eitanSlideEnds = ends;
window.__USA_PRESENTATION_TIMELINE__ = durations.map((duration, i) => ({
slide: i + 1,
base: eitanBaseForSlide(slides[i], i),
type: videoDurationsByIndex.has(i) ? 'video' : 'image',
mediaSeconds: Number(videoDurationsByIndex.get(i) || 0),
timelineSeconds: duration,
start: starts[i],
end: ends[i]
}));
}
eitanBuildTimeline();
function eitanManifestUrlForImage(img) {
const source = img.dataset.localBase || img.getAttribute('src') || '';
const slash = source.replace(/\\/g, '/').lastIndexOf('/');
if (slash < 0) return EITAN_MEDIA_MANIFEST_NAME;
return source.slice(0, slash + 1) + EITAN_MEDIA_MANIFEST_NAME;
}
function eitanReadInlineManifest() {
const node = document.getElementById('usa2027-inline-media-manifest');
if (!node) return { token:'missing', items:[] };
try {
const data = JSON.parse(node.textContent || '{}');
return { token:String(data.token || 'unknown'), items:Array.isArray(data.items) ? data.items : [] };
} catch(e) {
return { token:'invalid', items:[] };
}
}
async function eitanReadManifest(url) {
const data = eitanReadInlineManifest();
const baseUrl = new URL('.', document.baseURI || window.location.href).href;
return { url:baseUrl, items:data.items.map(item => {
if (!item || !item.url) return item;
return {...item, file:item.url};
}) };
}
function eitanWaitVideoMetadata(video, timeoutMs = 10000) {
if (Number.isFinite(video.duration) && video.duration > 0) return Promise.resolve(video.duration);
return new Promise((resolve, reject) => {
let timer;
const done = () => {
clearTimeout(timer);
video.removeEventListener('loadedmetadata', onReady);
video.removeEventListener('durationchange', onReady);
video.removeEventListener('error', onError);
};
const onReady = () => {
if (Number.isFinite(video.duration) && video.duration > 0) {
const value = video.duration;
done();
resolve(value);
}
};
const onError = () => { done(); reject(new Error(`וידאו לא נטען: ${video.dataset.mediaBase || ''}`)); };
timer = setTimeout(() => { done(); reject(new Error(`תם הזמן לטעינת נתוני הווידאו: ${video.dataset.mediaBase || ''}`)); }, timeoutMs);
video.addEventListener('loadedmetadata', onReady);
video.addEventListener('durationchange', onReady);
video.addEventListener('error', onError);
try { video.load(); } catch(e) {}
});
}
async function preparePresentationMedia() {
if (eitanMediaPrepared) return true;
if (eitanMediaPreparePromise) return eitanMediaPreparePromise;
eitanMediaPreparePromise = (async () => {
const images = [...document.querySelectorAll('img.hero-image')];
const manifestResults = [await eitanReadManifest('inline')];
const itemByBase = new Map();
manifestResults.forEach(result => {
result.items.forEach(item => {
if (!item || !item.base || item.type !== 'video' || !item.file) return;
const assetUrl = new URL(item.url || item.file, document.baseURI || window.location.href).href;
itemByBase.set(String(item.base).toLowerCase(), { ...item, assetUrl });
});
});
const videoDurationsByIndex = new Map();
for (let i = 0; i < slides.length; i++) {
const slide = slides[i];
const img = slide.querySelector('img.hero-image');
if (!img) continue;
const base = eitanBaseForSlide(slide, i);
const item = itemByBase.get(String(base).toLowerCase());
if (!item) continue;
const video = document.createElement('video');
video.className = 'hero-image hero-video';
video.muted = true;
video.defaultMuted = true;
video.autoplay = false;
video.loop = false;
video.controls = false;
video.playsInline = true;
video.setAttribute('playsinline', '');
video.setAttribute('webkit-playsinline', '');
video.preload = 'metadata';
video.dataset.mediaBase = base;
video.dataset.mediaSource = item.assetUrl;
video.setAttribute('aria-label', img.getAttribute('alt') || base);
let selectedSrc = item.assetUrl;
try {
if (typeof getCachedObjectUrl === 'function') {
const cached = await getCachedObjectUrl(item.assetUrl);
if (cached) selectedSrc = cached;
}
} catch(e) {}
video.src = selectedSrc;
img.replaceWith(video);
let duration = Number(item.duration || 0);
try {
const actual = await eitanWaitVideoMetadata(video, 10000);
if (Number.isFinite(actual) && actual > 0) duration = actual;
} catch(e) {
if (!(Number.isFinite(duration) && duration > 0)) throw e;
}
if (!(Number.isFinite(duration) && duration > 0)) {
throw new Error(`משך הווידאו לא תקין בשקף ${i + 1}.`);
}
eitanVideoDurations.set(i, duration);
videoDurationsByIndex.set(i, duration);
}
eitanBuildTimeline(videoDurationsByIndex);
eitanMediaPrepared = true;
window.__USA_PRESENTATION_MEDIA_READY__ = true;
return true;
})();
try {
return await eitanMediaPreparePromise;
} catch (e) {
eitanMediaPreparePromise = null;
throw e;
}
}
function pauseAllPresentationVideos(except = null) {
document.querySelectorAll('video.hero-video').forEach(video => {
if (video !== except) {
try { video.pause(); } catch(e) {}
}
});
}
function syncPresentationVideo(globalSeconds) {
if (!eitanMediaPrepared) return;
const activeIndex = timeToSlide(globalSeconds);
document.querySelectorAll('video.hero-video').forEach(video => {
const slide = video.closest('.slide');
const slideIndex = slides.indexOf(slide);
if (slideIndex !== activeIndex) {
try { video.pause(); } catch(e) {}
return;
}
const mediaDuration = Number(eitanVideoDurations.get(slideIndex) || video.duration || 0);
const local = Math.max(0, globalSeconds - Number(eitanSlideStarts[slideIndex] || 0));
const playbackPoint = Math.max(0, Math.min(Math.max(0, mediaDuration - 0.04), local));
try {
if (Math.abs((video.currentTime || 0) - playbackPoint) > 0.45) video.currentTime = playbackPoint;
if (local < mediaDuration - 0.04 && playing) {
const promise = video.play();
if (promise && typeof promise.catch === 'function') promise.catch(() => {});
} else {
video.pause();
if (mediaDuration > 0 && local >= mediaDuration - 0.04) video.currentTime = Math.max(0, mediaDuration - 0.04);
}
} catch(e) {}
});
}
/* EITAN_VIDEO_MEDIA_RUNTIME_END */
""".rstrip()
TIMING_FUNCTIONS_PATCH = r"""
function slideDurationInPart(part) {
const indices = [];
for (let i = part.start; i <= part.end; i++) indices.push(i);
return indices.reduce((sum, i) => sum + Number(eitanSlideTimelineDurations[i] || 0), 0) / Math.max(1, indices.length);
}
function slideToGlobalTime(slideIndex) {
const safe = Math.max(0, Math.min(slides.length - 1, Number(slideIndex) || 0));
return Number(eitanSlideStarts[safe] || 0);
}
function timeToSlide(seconds) {
const target = Math.max(0, Math.min(totalDuration, Number(seconds) || 0));
let low = 0;
let high = eitanSlideEnds.length - 1;
while (low < high) {
const mid = Math.floor((low + high) / 2);
if (target < eitanSlideEnds[mid] - 0.0005) high = mid;
else low = mid + 1;
}
return Math.max(0, Math.min(slides.length - 1, low));
}
""".rstrip()
def _replace_once(html: str, pattern: str, replacement: str, description: str, flags: int = 0) -> str:
updated, count = re.subn(pattern, replacement, html, count=1, flags=flags)
if count != 1:
raise MediaError(f"לא הצלחתי לעדכן את {description} בקובץ המצגת (נמצאו {count} התאמות).")
return updated
def patch_presentation_html(html: str) -> Tuple[str, bool]:
if "EITAN_VIDEO_MEDIA_RUNTIME_START" in html:
return html, False
# CSS
if "</head>" not in html.lower():
raise MediaError("קובץ המצגת חסר תגית </head>.")
html = re.sub(r"</head>", f"<style>\n{CSS_PATCH}\n</style>\n</head>", html, count=1, flags=re.I)
# Insert runtime immediately after totalDuration so all timing functions can use it.
total_pattern = r"(\bconst\s+totalDuration\s*=\s*parts\.reduce\([^;]+;\s*)"
html = _replace_once(
html,
total_pattern,
lambda m: m.group(1) + "\n" + JS_RUNTIME_PATCH + "\n",
"מנוע המדיה",
flags=re.S,
)
timing_pattern = (
r"\s*function\s+slideDurationInPart\s*\(part\)\s*\{.*?\}\s*"
r"function\s+slideToGlobalTime\s*\(slideIndex\)\s*\{.*?\}\s*"
r"function\s+timeToSlide\s*\(seconds\)\s*\{.*?\}\s*"
r"(?=function\s+getAudioDrivenSeconds)"
)
html = _replace_once(html, timing_pattern, "\n" + TIMING_FUNCTIONS_PATCH + "\n\n ", "פונקציות הסנכרון", flags=re.S)
# Start media preparation without awaiting it. This is intentional: the existing iPhone/Safari
# audio priming code must remain in the same user-gesture task. Awaiting video metadata before
# audio.play() would cause Safari to block the music.
start_anchor = r"(\s*await\s+applyCachedAudioSources\(\);)"
start_code = r"""\1
let eitanMediaStartError = null;
setLoader(3, 'בודק תמונות וסרטונים…');
const eitanMediaStartPromise = preparePresentationMedia().catch(e => {
eitanMediaStartError = e;
return false;
});"""
html = _replace_once(html, start_anchor, start_code, "התחלת טעינת המדיה")
# Join the existing preload barrier only after the original audio-prime calls have run.
preload_pattern = r"(await\s+Promise\.allSettled\(\[)(.*?)(\]\);)"
preload_replacement = r"\1\n eitanMediaStartPromise,\2\3"
html = _replace_once(html, preload_pattern, preload_replacement, "שילוב טעינת המדיה במחסום הטעינה", flags=re.S)
media_error_check = r"""
if (eitanMediaStartError) {
playing = false;
musicEnabled = false;
pauseAllAudio();
pauseAllPresentationVideos();
const message = eitanMediaStartError && eitanMediaStartError.message
? eitanMediaStartError.message
: String(eitanMediaStartError || 'כשל מדיה');
window.__USA_PRESENTATION_MEDIA_ERROR__ = message;
setLoader(0, 'שגיאת מדיה: ' + message);
if (startWithMusic) {
startWithMusic.disabled = false;
startWithMusic.textContent = '⚠️ בדוק את קבצי המדיה';
}
showToast(message);
return;
}
"""
preload_done_pattern = r"(await\s+Promise\.allSettled\(\[.*?\]\);)"
html = _replace_once(html, preload_done_pattern, lambda m: m.group(1) + media_error_check, "בדיקת תוצאת טעינת המדיה", flags=re.S)
# Keep video aligned on every clock frame.
sync_anchor = r"(if\s*\(nextPart\s*!==\s*currentPartIndex\)\s*\{\s*alignAudioToTime\(seconds,\s*musicEnabled\)\.catch\(\(\)\s*=>\s*\{\}\);\s*\})"
html = _replace_once(html, sync_anchor, r"\1\n syncPresentationVideo(seconds);", "סנכרון הסרטון", flags=re.S)
# Manual seek must also update the video while paused.
seek_pattern = r"(function\s+seekToSlide\s*\(slideIndex\)\s*\{.*?\bupdate\(\);)"
html = _replace_once(html, seek_pattern, r"\1\n syncPresentationVideo(targetTime);", "סנכרון בעת מעבר ידני", flags=re.S)
# Pause and finish stop videos too.
pause_pattern = r"(function\s+pausePresentation\s*\(\)\s*\{.*?pauseAllAudio\(\);)"
html = _replace_once(html, pause_pattern, r"\1\n pauseAllPresentationVideos();", "עצירת סרטונים בהשהיה", flags=re.S)
finish_pattern = r"(function\s+finishPresentation\s*\(\)\s*\{.*?pauseAllAudio\(\);)"
html = _replace_once(html, finish_pattern, r"\1\n pauseAllPresentationVideos();", "עצירת סרטונים בסיום", flags=re.S)
# Offline cache: prepare DOM first and include videos + manifests.
cache_prepare_pattern = r"(async\s+function\s+prepareOfflineCache\s*\(\)\s*\{.*?try\s*\{\s*if\s*\(navigator\.storage.*?\}\s*catch\(e\)\s*\{\}\s*)(const\s+urls\s*=\s*allOfflineAssetUrls\(\);)"
html = _replace_once(
html,
cache_prepare_pattern,
r"\1try { await preparePresentationMedia(); } catch(e) { setOfflineStatus(e.message || String(e)); prepareOfflineCacheBtn.disabled = false; return; }\n \2",
"הכנת סרטונים לשמירה אופליין",
flags=re.S,
)
offline_pattern = r"function\s+allOfflineAssetUrls\s*\(\)\s*\{.*?\n\s*\}"
offline_replacement = r"""function allOfflineAssetUrls() {
// מקור אמת יחיד: HTML המצגת + רשימת המדיה המוטמעת.
// אין לסרוק את ה-DOM בנוסף, אחרת תמונה ישנה שהוחלפה בסרטון
// נספרת לצד הסרטון ונוצר פער שקרי בין קובץ הטיול למצגת.
const inline = eitanReadInlineManifest();
const declared = inline.items
.map(item => item && (item.url || item.file))
.filter(src => src && !/^data:/i.test(src))
.map(absoluteAssetUrl);
return [...new Set([canonicalPresentationUrl(), ...declared])];
}"""
html = _replace_once(html, offline_pattern, offline_replacement, "רשימת המדיה לשמירה אופליין", flags=re.S)
# Human-facing loading wording only; no functional dependency.
html = html.replace("טוען תמונות ומוזיקה…", "טוען מדיה ומוזיקה…")
html = html.replace("מכין מוזיקה וטוען תמונות…", "מכין מוזיקה וטוען מדיה…")
html = html.replace("טוען תמונות ראשונות…", "טוען מדיה ראשונה…")
return html, True
def patch_presentation_file(path: Path, *, backup: bool = True) -> bool:
path = Path(path)
if not path.is_file():
raise MediaError(f"קובץ המצגת לא נמצא: {path}")
original = path.read_text(encoding="utf-8")
updated, changed = patch_presentation_html(original)
if not changed:
return False
if backup:
backup_dir = path.parent / ".admin_backups"
backup_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
shutil.copy2(path, backup_dir / f"{path.name}.{stamp}.bak")
temp = path.with_suffix(path.suffix + ".tmp")
temp.write_text(updated, encoding="utf-8")
# Re-inspect before replacing to ensure core data survived.
inspect_presentation_html(updated)
os.replace(temp, path)
return True
def write_timeline_report(presentation_path: Path, media_dir: Path, output_path: Optional[Path] = None) -> Path:
info = inspect_presentation_file(presentation_path)
videos = video_duration_map(media_dir, info.slide_bases)
timeline = allocate_timeline(info.slide_bases, info.parts, videos)
if output_path is None:
output_path = Path(media_dir) / "media-timeline-report.json"
payload = {
"patchVersion": PATCH_VERSION,
"presentation": str(presentation_path),
"mediaDirectory": str(media_dir),
"totalSeconds": timeline.total_seconds,
"warnings": list(timeline.warnings),
"entries": [asdict(entry) for entry in timeline.entries],
}
Path(output_path).write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
return Path(output_path)
# ================= END USA VIDEO MEDIA ENGINE =================
def normalize_usa_image_target(target_variant):
target_variant = (target_variant or "phone").strip().lower()
if target_variant in {"mobile", "iphone", "phone", "פלאפון", "טלפון"}:
return "phone"
if target_variant in {"tv", "television", "screen", "טלוויזיה", "טלויזיה"}:
return "tv"
if target_variant not in USA_IMAGE_TARGETS:
raise ValueError("יעד מדיה למצגת ארה״ב לא תקין. יש לבחור פלאפון או טלוויזיה.")
return target_variant
def usa_image_target_info(target_variant):
target_variant = normalize_usa_image_target(target_variant)
return target_variant, USA_IMAGE_TARGETS[target_variant]
def usa_image_dir_for_target(target_variant):
target_variant, info = usa_image_target_info(target_variant)
return os.path.join(usa_folder_for_image_target(target_variant), info["folder"])
def ensure_usa_presentation_uses_image_target(target_variant):
"""Make sure each USA presentation points to the media folder that exists in its own root."""
target_variant, info = usa_image_target_info(target_variant)
presentation_key = info.get("presentation_key")
filename = USA_PRESENTATION_FILES.get(presentation_key)
if not filename:
return {"updated": False, "reason": "presentation_key_not_found", "target": target_variant}
path = os.path.join(usa_folder_for_presentation_key(presentation_key), filename)
if not os.path.isfile(path):
return {"updated": False, "reason": "presentation_file_missing", "target": target_variant, "path": path}
html = read_text_file_utf8(path)
desired = info["rel_prefix"] + "/"
new_html = html
# Old experimental TV path and the accidental doubled media path are both invalid for the USA_TV folder shown on disk.
for wrong in ("media/usa-presentation/images-tv/", "media/usa-presentation/media/images/"):
new_html = new_html.replace(wrong, desired)
if new_html == html:
return {"updated": False, "reason": "already_correct", "target": target_variant, "path": path}
backup = backup_existing_file(path, f"USA_presentation_image_path_{target_variant}")
write_text_file_utf8(path, new_html)
return {"updated": True, "target": target_variant, "path": path, "backup": backup, "sha256": file_sha256(path)}
def usa_presentation_path_for_target(target_variant):
target_variant, info = usa_image_target_info(target_variant)
presentation_key = info.get("presentation_key")
filename = USA_PRESENTATION_FILES.get(presentation_key)
if not filename:
raise ValueError("לא נמצא מיפוי קובץ מצגת ליעד המדיה.")
return os.path.join(usa_folder_for_presentation_key(presentation_key), filename)
def usa_media_files_for_target(target_variant):
return media_files_by_base(Path(usa_image_dir_for_target(target_variant)))
def probe_video_duration_seconds(path):
probe = probe_audio_duration_seconds(path)
if probe.get("ok"):
value = float(probe.get("duration_seconds") or 0)
if value > 0:
return value
return mp4_duration_seconds(Path(path))
def probe_video_stream_info(path):
ffprobe = find_ffprobe_executable()
if not ffprobe:
return {"ok": False, "codec": "unknown", "width": 0, "height": 0, "ffprobe": ""}
cmd = [
ffprobe, "-v", "error", "-select_streams", "v:0",
"-show_entries", "stream=codec_name,width,height,pix_fmt",
"-of", "json", path,
]
try:
cp = run_hidden_subprocess(cmd, timeout=30)
if cp.returncode != 0:
return {"ok": False, "codec": "unknown", "error": (cp.stderr or cp.stdout or "")[-1000:], "ffprobe": ffprobe}
data = json.loads(cp.stdout or "{}")
stream = (data.get("streams") or [{}])[0]
return {
"ok": bool(stream),
"codec": str(stream.get("codec_name") or "unknown").lower(),
"width": int(stream.get("width") or 0),
"height": int(stream.get("height") or 0),
"pix_fmt": str(stream.get("pix_fmt") or ""),
"ffprobe": ffprobe,
}
except Exception as exc:
return {"ok": False, "codec": "unknown", "error": repr(exc), "ffprobe": ffprobe}
def normalize_video_to_mp4(source_path, target_path, original_name=""):
"""Create a browser-safe MP4. Preserve H.264 by remuxing; transcode only when required."""
ensure_parent_dir(target_path)
size = os.path.getsize(source_path) if os.path.exists(source_path) else 0
if size <= 0:
raise ValueError("קובץ הסרטון ריק.")
if size > USA_VIDEO_MAX_SINGLE_FILE_BYTES:
raise ValueError(f"קובץ הסרטון גדול מדי: {format_bytes(size)}. מגבלה: {format_bytes(USA_VIDEO_MAX_SINGLE_FILE_BYTES)}")
info = probe_video_stream_info(source_path)
ffmpeg = find_ffmpeg_executable()
tmp_target = target_path + ".video.tmp.mp4"
mode = "copied_mp4"
if ffmpeg:
codec = info.get("codec") or "unknown"
if codec == "h264":
cmd = [ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-i", source_path,
"-map", "0:v:0", "-an", "-c:v", "copy", "-movflags", "+faststart", tmp_target]
mode = "remuxed_h264_faststart"
else:
cmd = [ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-i", source_path,
"-map", "0:v:0", "-an", "-c:v", "libx264", "-preset", "medium", "-crf", "20",
"-pix_fmt", "yuv420p", "-movflags", "+faststart", tmp_target]
mode = "transcoded_h264_faststart"
cp = run_hidden_subprocess(cmd, timeout=600)
if cp.returncode != 0 or not os.path.exists(tmp_target) or os.path.getsize(tmp_target) <= 0:
try:
if os.path.exists(tmp_target):
os.remove(tmp_target)
except Exception:
pass
raise ValueError("FFmpeg לא הצליח להכין MP4 תקין: " + ((cp.stderr or cp.stdout or "שגיאה לא ידועה")[-1600:]))
os.replace(tmp_target, target_path)
else:
# MP4/M4V are both ISO-BMFF containers. Validate structure before copying.
mp4_duration_seconds(Path(source_path))
tmp_copy = target_path + ".copy.tmp"
shutil.copy2(source_path, tmp_copy)
os.replace(tmp_copy, target_path)
mode = "copied_validated_mp4_without_ffmpeg"
duration = probe_video_duration_seconds(target_path)
if duration <= 0:
raise ValueError("לא זוהה משך תקין לסרטון לאחר השמירה.")
final_info = probe_video_stream_info(target_path)
return {
"mode": mode,
"original_size": size,
"final_size": os.path.getsize(target_path),
"saved_bytes": max(0, size - os.path.getsize(target_path)),
"duration_seconds": duration,
"duration_display": format_seconds(duration),
"codec": final_info.get("codec") or info.get("codec") or "unknown",
"width": final_info.get("width") or info.get("width") or 0,
"height": final_info.get("height") or info.get("height") or 0,
"ffmpeg": ffmpeg or "not_found",
}
def usa_media_sibling_paths(media_dir, media_key):
return [os.path.join(media_dir, media_key + ext) for ext in sorted(USA_MEDIA_EXTENSIONS)]
def calculate_usa_timeline_for_html(target_variant, html_text):
"""Calculate a prospective USA timeline against current media without writing files."""
target_variant, _target_info = usa_image_target_info(target_variant)
info = inspect_presentation_html(html_text)
allowed = {base.lower() for base in info.slide_bases}
media_map = media_files_by_base(Path(usa_image_dir_for_target(target_variant)))
videos = {}
for base, path in media_map.items():
if base in allowed and Path(path).suffix.lower() in USA_VIDEO_EXTENSIONS:
videos[base] = probe_video_duration_seconds(str(path))
timeline = allocate_timeline(info.slide_bases, info.parts, videos)
return info, timeline, videos
def validate_usa_media_prospective(target_variant, overrides=None):
target_variant, target_info = usa_image_target_info(target_variant)
presentation_path = usa_presentation_path_for_target(target_variant)
if not os.path.isfile(presentation_path):
raise FileNotFoundError(f"קובץ המצגת לא נמצא ולכן אי אפשר לאמת סנכרון: {presentation_path}")
info = inspect_presentation_file(Path(presentation_path))
allowed = {base.lower() for base in info.slide_bases}
media_map = media_files_by_base(Path(usa_image_dir_for_target(target_variant)))
for key, path in (overrides or {}).items():
k = str(key).lower()
if k not in allowed:
raise ValueError(f"שם המדיה {key} אינו תואם לשקף במצגת.")
if path is None:
media_map.pop(k, None)
else:
media_map[k] = Path(path)
videos = {}
for base, path in media_map.items():
if base in allowed and Path(path).suffix.lower() in USA_VIDEO_EXTENSIONS:
videos[base] = probe_video_duration_seconds(str(path))
timeline = allocate_timeline(info.slide_bases, info.parts, videos)
return info, timeline, videos
def write_inline_usa_media_manifest(target_variant, info=None, videos=None):
"""Embed the authoritative media list inside the presentation HTML itself.
No external manifest or Service Worker is required. Both the trip page and the
presentation read this exact JSON block, so their file counts cannot diverge.
"""
target_variant, target_info = usa_image_target_info(target_variant)
presentation_path = usa_presentation_path_for_target(target_variant)
if not os.path.isfile(presentation_path):
return {"ok": False, "reason": "presentation_missing", "path": presentation_path}
if info is None or videos is None:
info, _timeline, videos = validate_usa_media_prospective(target_variant)
media_map = media_files_by_base(Path(usa_image_dir_for_target(target_variant)))
rel_prefix = target_info.get("rel_prefix") or "media/usa-presentation/images"
items = []
for idx, base in enumerate(info.slide_bases):
path = media_map.get(str(base).lower())
if path is None:
# Keep the presentation's conventional JPG path as the declared image.
url = f"{rel_prefix}/{base}.jpg"
items.append({"slide": idx + 1, "base": base, "type": "image", "url": url})
continue
ext = Path(path).suffix.lower()
item = {"slide": idx + 1, "base": base, "type": "video" if ext in USA_VIDEO_EXTENSIONS else "image", "url": f"{rel_prefix}/{Path(path).name}"}
if ext in USA_VIDEO_EXTENSIONS:
item["duration"] = round(float(videos.get(str(base).lower()) or probe_video_duration_seconds(str(path))), 6)
items.append(item)
items.extend([
{"type":"audio","base":"usa-roadtrip-theme","url":"media/usa-presentation/music/usa-roadtrip-theme.mp3"},
{"type":"audio","base":"usa-part2-theme","url":"media/usa-presentation/music/usa-part2-theme.mp3"},
])
canonical = json.dumps(items, ensure_ascii=False, separators=(",", ":"))
token = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
payload = json.dumps({"schema":1,"token":token,"items":items}, ensure_ascii=False, separators=(",", ":"))
script = f'<script id="usa2027-inline-media-manifest" type="application/json">{payload}</script>'
html = read_text_file_utf8(presentation_path)
rx = re.compile(r'<script\b[^>]*\bid=["\']usa2027-inline-media-manifest["\'][^>]*>[\s\S]*?</script>', re.I)
if rx.search(html):
updated = rx.sub(script, html, count=1)
else:
updated = re.sub(r'(<script\b)', script + r'\n\1', html, count=1, flags=re.I)
if updated != html:
write_text_file_utf8_atomic_with_backup(presentation_path, updated, "USA_inline_media_manifest")
verify = read_text_file_utf8(presentation_path)
if payload not in verify:
raise RuntimeError("האימות נכשל: רשימת המדיה המוטמעת לא נשמרה במצגת.")
return {"ok":True,"path":presentation_path,"token":token,"count":len(items),"sha256":file_sha256(presentation_path)}
def write_usa_media_metadata(target_variant):
target_variant, target_info = usa_image_target_info(target_variant)
presentation_path = usa_presentation_path_for_target(target_variant)
media_dir = usa_image_dir_for_target(target_variant)
if not os.path.isfile(presentation_path):
return {"ok": False, "reason": "presentation_missing", "path": presentation_path}
info, timeline, videos = validate_usa_media_prospective(target_variant)
manifest_path = write_manifest(Path(media_dir), info.slide_bases)
report_path = write_timeline_report(Path(presentation_path), Path(media_dir), Path(media_dir) / USA_MEDIA_TIMELINE_REPORT_NAME)
inline_manifest = write_inline_usa_media_manifest(target_variant, info, videos)
return {
"ok": True,
"target": target_variant,
"manifest": str(manifest_path),
"inline_manifest": inline_manifest,
"report": str(report_path),
"video_count": len(videos),
"total_seconds": timeline.total_seconds,
"warnings": list(timeline.warnings),
}
def ensure_usa_presentation_media_engine(target_variant):
target_variant, target_info = usa_image_target_info(target_variant)
path_update = ensure_usa_presentation_uses_image_target(target_variant)
presentation_path = usa_presentation_path_for_target(target_variant)
if not os.path.isfile(presentation_path):
raise FileNotFoundError(presentation_path)
changed = patch_presentation_file(Path(presentation_path), backup=True)
metadata = write_usa_media_metadata(target_variant)
return {
"target": target_variant,
"presentation": presentation_path,
"patched_now": bool(changed),
"patch_version": USA_MEDIA_PATCH_VERSION,
"path_update": path_update,
"metadata": metadata,
}
def refresh_usa_media_after_timing_change(target_variant):
"""Keep reports current after PART_1_SECONDS/PART_2_SECONDS changes without changing music logic."""
try:
presentation_path = usa_presentation_path_for_target(target_variant)
if not os.path.isfile(presentation_path):
return {"ok": False, "reason": "presentation_missing"}
html = read_text_file_utf8(presentation_path)
has_videos = any(Path(p).suffix.lower() in USA_VIDEO_EXTENSIONS for p in usa_media_files_for_target(target_variant).values())
if "EITAN_VIDEO_MEDIA_RUNTIME_START" not in html and not has_videos:
return {"ok": True, "skipped": True, "reason": "no_video_media"}
return ensure_usa_presentation_media_engine(target_variant)
except Exception as exc:
return {"ok": False, "error": repr(exc)}
def _restore_media_transaction(media_dir, affected_keys, rollback_dir):
for key in affected_keys:
for path in usa_media_sibling_paths(media_dir, key):
try:
if os.path.exists(path):
os.remove(path)
except Exception:
pass
if os.path.isdir(rollback_dir):
for name in os.listdir(rollback_dir):
src = os.path.join(rollback_dir, name)
if os.path.isfile(src):
shutil.copy2(src, os.path.join(media_dir, name))
def save_usa_presentation_media(media_key, uploaded_file, target_variant="phone"):
target_variant, target_info = usa_image_target_info(target_variant)
if media_key not in USA_IMAGE_KEYS:
return None
original_name = uploaded_file.filename or ""
ext = file_ext(original_name)
if ext not in USA_MEDIA_EXTENSIONS:
raise ValueError("סיומת לא נתמכת. העלה תמונה או סרטון MP4/M4V.")
media_dir = usa_image_dir_for_target(target_variant)
os.makedirs(media_dir, exist_ok=True)
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
work_dir = tempfile.mkdtemp(prefix=f"usa_media_{target_variant}_{media_key}_", dir=ADMIN_UPLOAD_TMP_DIR)
rollback_dir = os.path.join(work_dir, "rollback")
os.makedirs(rollback_dir, exist_ok=True)
source_path = os.path.join(work_dir, secure_filename(original_name) or (media_key + ext))
uploaded_file.save(source_path)
uploaded_size = os.path.getsize(source_path)
if uploaded_size <= 0:
shutil.rmtree(work_dir, ignore_errors=True)
raise ValueError("קובץ המדיה ריק.")
if ext in USA_IMAGE_EXTENSIONS and uploaded_size > USA_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES:
shutil.rmtree(work_dir, ignore_errors=True)
raise ValueError(f"קובץ התמונה גדול מדי: {format_bytes(uploaded_size)}. מגבלה: {format_bytes(USA_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES)}")
try:
if ext in USA_VIDEO_EXTENSIONS:
prepared = os.path.join(work_dir, media_key + ".mp4")
media = normalize_video_to_mp4(source_path, prepared, original_name)
canonical_ext = ".mp4"
kind = "video"
else:
prepared = os.path.join(work_dir, media_key + ".jpg")
media = convert_or_copy_image_to_jpg(source_path, prepared, original_name)
canonical_ext = ".jpg"
kind = "image"
info, prospective_timeline, videos = validate_usa_media_prospective(target_variant, {media_key: prepared})
for sibling in usa_media_sibling_paths(media_dir, media_key):
if os.path.isfile(sibling):
shutil.copy2(sibling, os.path.join(rollback_dir, os.path.basename(sibling)))
backup_existing_file(sibling, target_info["backup_tag"] + "_media_single")
target_path = os.path.join(media_dir, media_key + canonical_ext)
for sibling in usa_media_sibling_paths(media_dir, media_key):
if os.path.exists(sibling):
os.remove(sibling)
tmp_target = target_path + ".tmp"
shutil.copy2(prepared, tmp_target)
os.replace(tmp_target, target_path)
engine = ensure_usa_presentation_media_engine(target_variant)
return {
"target": target_variant,
"target_label": target_info["label"],
"key": media_key,
"kind": kind,
"rel": os.path.join(target_info["rel_prefix"], os.path.basename(target_path)).replace(os.sep, "/"),
"path": target_path,
"mode": media.get("mode", kind),
"media": media,
"sha256": file_sha256(target_path),
"size": os.path.getsize(target_path),
"video_count": len(videos),
"timeline_total_seconds": prospective_timeline.total_seconds,
"timeline_warnings": list(prospective_timeline.warnings),
"presentation_media_engine": engine,
}
except Exception:
_restore_media_transaction(media_dir, [media_key], rollback_dir)
try:
write_usa_media_metadata(target_variant)
except Exception:
pass
raise
finally:
shutil.rmtree(work_dir, ignore_errors=True)
def save_usa_presentation_image(image_key, uploaded_file, target_variant="phone"):
"""Backward-compatible name: now accepts either an image or MP4/M4V video."""
return save_usa_presentation_media(image_key, uploaded_file, target_variant)
def save_usa_presentation_media_zip(uploaded_file, target_variant="phone"):
target_variant, target_info = usa_image_target_info(target_variant)
media_dir = usa_image_dir_for_target(target_variant)
os.makedirs(media_dir, exist_ok=True)
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
work_dir = tempfile.mkdtemp(prefix=f"usa_media_zip_{target_variant}_", dir=ADMIN_UPLOAD_TMP_DIR)
zip_path = os.path.join(work_dir, "upload.zip")
extract_dir = os.path.join(work_dir, "extract")
prepared_dir = os.path.join(work_dir, "prepared")
rollback_dir = os.path.join(work_dir, "rollback")
os.makedirs(extract_dir, exist_ok=True)
os.makedirs(prepared_dir, exist_ok=True)
os.makedirs(rollback_dir, exist_ok=True)
uploaded_file.save(zip_path)
if os.path.getsize(zip_path) > USA_MEDIA_ZIP_MAX_BYTES:
shutil.rmtree(work_dir, ignore_errors=True)
raise ValueError(f"קובץ ZIP גדול מהמגבלה: {format_bytes(USA_MEDIA_ZIP_MAX_BYTES)}")
imported = []
skipped = []
prepared = {}
affected = []
try:
with zipfile.ZipFile(zip_path, "r") as zf:
infos = [i for i in zf.infolist() if not i.is_dir()]
if len(infos) > USA_MEDIA_ZIP_MAX_FILES:
raise ValueError(f"יותר מדי קבצים ב־ZIP: {len(infos)}")
total = sum(max(0, i.file_size) for i in infos)
if total > USA_MEDIA_ZIP_MAX_UNCOMPRESSED_BYTES:
raise ValueError("הגודל לאחר חילוץ גדול מהמגבלה.")
seen = set()
for info in infos:
raw_name = (info.filename or "").replace("\\", "/")
base_name = os.path.basename(raw_name)
if not base_name or base_name.startswith(".") or raw_name.startswith("__MACOSX/"):
continue
stem, ext = os.path.splitext(base_name)
stem = stem.strip()
ext = ext.lower()
if ext not in USA_MEDIA_EXTENSIONS:
skipped.append(base_name + " — פורמט לא נתמך")
continue
if stem not in USA_IMAGE_KEYS:
skipped.append(base_name + " — השם אינו תואם לשקף")
continue
if stem in seen:
raise ValueError(f"נמצאה כפילות עבור השקף {stem} בתוך ה־ZIP.")
if info.file_size > USA_MEDIA_ZIP_MAX_SINGLE_FILE_BYTES:
skipped.append(base_name + " — גדול ממגבלת הקובץ הכללית")
continue
if ext in USA_VIDEO_EXTENSIONS and info.file_size > USA_VIDEO_MAX_SINGLE_FILE_BYTES:
skipped.append(base_name + " — הסרטון גדול מדי")
continue
if ext in USA_IMAGE_EXTENSIONS and info.file_size > USA_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES:
skipped.append(base_name + " — התמונה גדולה מדי")
continue
seen.add(stem)
source = os.path.join(extract_dir, stem + ext)
with zf.open(info, "r") as src, open(source, "wb") as dst:
shutil.copyfileobj(src, dst)
if ext in USA_VIDEO_EXTENSIONS:
dest = os.path.join(prepared_dir, stem + ".mp4")
media = normalize_video_to_mp4(source, dest, base_name)
kind = "video"
else:
dest = os.path.join(prepared_dir, stem + ".jpg")
media = convert_or_copy_image_to_jpg(source, dest, base_name)
kind = "image"
prepared[stem] = (dest, kind, media)
affected.append(stem)
if not prepared:
raise ValueError("לא נמצאו ב־ZIP קובצי תמונה או וידאו עם שמות שקפים תקינים.")
overrides = {key: item[0] for key, item in prepared.items()}
info, prospective_timeline, videos = validate_usa_media_prospective(target_variant, overrides)
for key in affected:
for sibling in usa_media_sibling_paths(media_dir, key):
if os.path.isfile(sibling):
shutil.copy2(sibling, os.path.join(rollback_dir, os.path.basename(sibling)))
backup_existing_file(sibling, target_info["backup_tag"] + "_media_zip")
for key, (prepared_path, kind, media) in prepared.items():
for sibling in usa_media_sibling_paths(media_dir, key):
if os.path.exists(sibling):
os.remove(sibling)
target = os.path.join(media_dir, os.path.basename(prepared_path))
tmp_target = target + ".tmp"
shutil.copy2(prepared_path, tmp_target)
os.replace(tmp_target, target)
if kind == "video":
imported.append(f"{os.path.basename(target)} — סרטון {media.get('duration_display','')} ({format_bytes(media.get('final_size',0))})")
else:
imported.append(f"{os.path.basename(target)} — תמונה {format_bytes(media.get('original_size',0))} → {format_bytes(media.get('final_size',0))}")
engine = ensure_usa_presentation_media_engine(target_variant)
existing = media_files_by_base(Path(media_dir))
missing = sorted(key for key in USA_IMAGE_KEYS if key.lower() not in existing)
return imported, skipped, missing, {
"target": target_variant,
"target_label": target_info["label"],
"video_count": len(videos),
"timeline_total_seconds": prospective_timeline.total_seconds,
"timeline_warnings": list(prospective_timeline.warnings),
"presentation_media_engine": engine,
}
except Exception:
_restore_media_transaction(media_dir, affected, rollback_dir)
try:
write_usa_media_metadata(target_variant)
except Exception:
pass
raise
finally:
shutil.rmtree(work_dir, ignore_errors=True)
def save_usa_presentation_images_zip(uploaded_file, target_variant="phone"):
"""Backward-compatible name: ZIP may now contain images and MP4/M4V videos."""
return save_usa_presentation_media_zip(uploaded_file, target_variant)
for folder_path in FOLDERS.values():
os.makedirs(folder_path, exist_ok=True)
os.makedirs(os.path.join(FOLDERS["USA"], "media", "usa-presentation", "music"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["USA"], "media", "usa-presentation", "images"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["USA_TV"], "media", "usa-presentation", "music"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["USA_TV"], "media", "usa-presentation", "images"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["JAPAN"], "media", "images"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["JAPAN"], "media", "japan-experience", "music"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["JAPAN"], "media", "japan-flight", "music"), exist_ok=True)
os.makedirs(os.path.join(FOLDERS["JAPAN"], "media", "music"), exist_ok=True)
os.makedirs(ADMIN_BACKUP_DIR, exist_ok=True)
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
def no_cache_response(response):
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0, private"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
# Prevent old HTML/images from being reused after upload when the browser sends If-None-Match/If-Modified-Since.
response.headers.pop("ETag", None)
response.headers.pop("Last-Modified", None)
return response
@app.after_request
def add_global_no_cache_headers(response):
return no_cache_response(response)
def _safe_static_path(directory, file_name):
root = os.path.abspath(directory)
candidate = os.path.abspath(os.path.join(root, file_name))
if candidate != root and not candidate.startswith(root + os.sep):
raise ValueError("נתיב קובץ לא מורשה")
return candidate
def _parse_single_http_range(value, size):
if not value or not value.startswith("bytes="):
return None
spec = value[6:].split(",", 1)[0].strip()
if "-" not in spec:
raise ValueError("טווח HTTP לא תקין")
left, right = spec.split("-", 1)
if left == "":
suffix = int(right)
if suffix <= 0:
raise ValueError("טווח suffix לא תקין")
start = max(0, size - suffix)
end = size - 1
else:
start = int(left)
end = int(right) if right else size - 1
if start < 0 or start >= size or end < start:
raise ValueError("טווח מחוץ לקובץ")
return start, min(end, size - 1)
def _stream_file_range(path, start, length, chunk_size=1024 * 1024):
remaining = length
with open(path, "rb") as fh:
fh.seek(start)
while remaining > 0:
chunk = fh.read(min(chunk_size, remaining))
if not chunk:
break
remaining -= len(chunk)
yield chunk
def serve_file_no_cache(directory, file_name):
try:
full_path = _safe_static_path(directory, file_name)
if not os.path.isfile(full_path):
return "File not found", 404
ext = file_ext(full_path)
if ext in USA_VIDEO_EXTENSIONS:
size = os.path.getsize(full_path)
content_type = mimetypes.guess_type(full_path)[0] or "video/mp4"
range_header = request.headers.get("Range", "")
try:
parsed = _parse_single_http_range(range_header, size) if range_header else None
except Exception:
response = Response(status=416)
response.headers["Content-Range"] = f"bytes */{size}"
response.headers["Accept-Ranges"] = "bytes"
return no_cache_response(response)
if parsed:
start, end = parsed
length = end - start + 1
body = () if request.method == "HEAD" else _stream_file_range(full_path, start, length)
response = Response(body, status=206, mimetype=content_type, direct_passthrough=True)
response.headers["Content-Range"] = f"bytes {start}-{end}/{size}"
response.headers["Content-Length"] = str(length)
response.headers["Accept-Ranges"] = "bytes"
return no_cache_response(response)
response = make_response(send_file(full_path, mimetype=content_type, conditional=True, as_attachment=False))
response.headers["Accept-Ranges"] = "bytes"
return no_cache_response(response)
try:
response = make_response(send_from_directory(directory, file_name, conditional=False, max_age=0))
except TypeError:
response = make_response(send_from_directory(directory, file_name))
return no_cache_response(response)
except ValueError:
return "Forbidden", 403
def file_sha256(path):
import hashlib
h = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def ensure_parent_dir(path):
parent = os.path.dirname(path)
if parent:
os.makedirs(parent, exist_ok=True)
def backup_existing_file(target_path, backup_group="uploads"):
if not os.path.exists(target_path) or not os.path.isfile(target_path):
return None
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_group = secure_filename(str(backup_group)) or "uploads"
safe_name = secure_filename(os.path.basename(target_path)) or "file"
backup_dir = os.path.join(UPLOAD_BACKUP_DIR, safe_group)
os.makedirs(backup_dir, exist_ok=True)
backup_path = os.path.join(backup_dir, f"{stamp}_{safe_name}")
shutil.copy2(target_path, backup_path)
return backup_path
def atomic_save_upload(uploaded_file, target_path, backup_group="uploads"):
"""Save uploaded file atomically: tmp -> validate size -> backup old -> os.replace."""
ensure_parent_dir(target_path)
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
safe_target = secure_filename(os.path.basename(target_path)) or "upload.bin"
tmp_path = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"tmp_{stamp}_{os.getpid()}_{safe_target}")
try:
uploaded_file.save(tmp_path)
size = os.path.getsize(tmp_path)
if size <= 0:
raise ValueError("uploaded file is empty")
backup_path = backup_existing_file(target_path, backup_group)
os.replace(tmp_path, target_path)
final_size = os.path.getsize(target_path)
if final_size != size:
raise IOError(f"size mismatch after save: tmp={size}, final={final_size}")
return {"path": target_path, "size": final_size, "backup": backup_path, "sha256": file_sha256(target_path)}
finally:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
def folder_key_for_path(folder_path):
try:
abs_path = os.path.abspath(folder_path)
for key, value in FOLDERS.items():
if os.path.abspath(value) == abs_path:
return key
except Exception:
pass
return None
def find_html_file(folder_path):
# Use explicit entry files first, especially for USA_TV where the folder can also contain
# usa2027-presentation-tv.html. The quick trip route must open the trip file, not the presentation.
if not os.path.isdir(folder_path):
return None
folder_key = folder_key_for_path(folder_path)
for preferred in APP_ENTRY_FILES.get(folder_key, []):
if os.path.exists(os.path.join(folder_path, preferred)):
return preferred
index_path = os.path.join(folder_path, "index.html")
if os.path.exists(index_path):
return "index.html"
html_files = sorted(f for f in os.listdir(folder_path) if f.lower().endswith((".html", ".htm")))
if html_files:
return html_files[0]
return None
def clean_wrong_main_files(folder_path):
wrong_names = {"html", "htm", "index", "index.htm", "download", "unknown"}
for name in os.listdir(folder_path):
full_path = os.path.join(folder_path, name)
if not os.path.isfile(full_path):
continue
if name.lower().strip() in wrong_names:
try:
os.remove(full_path)
except Exception:
pass
def preferred_main_upload_filename(folder):
# One authoritative filename per app root.
# Critical for USA_TV: the quick-open button and the /USA_TV/ route both expect the TV trip file,
# not index.html and not the TV presentation. This prevents upload/open path drift.
if folder == "USA_TV":
return "usa2027-tv.html"
return "index.html"
def save_uploaded_file(folder, uploaded_file):
folder_path = FOLDERS[folder]
os.makedirs(folder_path, exist_ok=True)
original_filename = uploaded_file.filename or ""
safe_name = secure_filename(original_filename)
if folder in APP_FOLDERS:
clean_wrong_main_files(folder_path)
filename = preferred_main_upload_filename(folder)
else:
filename = safe_name or f"upload_{int(time.time())}"
target_path = os.path.join(folder_path, filename)
result = atomic_save_upload(uploaded_file, target_path, backup_group=f"{folder}_main")
return filename, result
def save_japan_presentation(presentation_key, uploaded_file):
if presentation_key not in JAPAN_PRESENTATION_FILES:
return None
japan_folder = FOLDERS["JAPAN"]
os.makedirs(japan_folder, exist_ok=True)
filename = JAPAN_PRESENTATION_FILES[presentation_key]
target_path = os.path.join(japan_folder, filename)
result = atomic_save_upload(uploaded_file, target_path, backup_group="JAPAN_presentations")
return filename, result
def save_usa_presentation(presentation_key, uploaded_file):
if presentation_key not in USA_PRESENTATION_FILES:
return None
usa_folder = usa_folder_for_presentation_key(presentation_key)
os.makedirs(usa_folder, exist_ok=True)
filename = USA_PRESENTATION_FILES[presentation_key]
target_path = os.path.join(usa_folder, filename)
target_variant = "tv" if presentation_key == "usa_presentation_tv" else "phone"
result = atomic_save_upload(uploaded_file, target_path, backup_group="USA_TV_presentations" if presentation_key == "usa_presentation_tv" else "USA_presentations")
try:
engine = ensure_usa_presentation_media_engine(target_variant)
result["presentation_media_engine"] = engine
except Exception:
backup = result.get("backup")
try:
if backup and os.path.isfile(backup):
shutil.copy2(backup, target_path)
elif os.path.exists(target_path):
os.remove(target_path)
except Exception:
pass
raise
return filename, result
def save_usa_music(music_key, uploaded_file):
if music_key not in USA_MUSIC_FILES:
return None
rel_path = USA_MUSIC_FILES[music_key]
result = save_audio_upload_to_absolute_targets(uploaded_file, usa_music_absolute_targets(rel_path), backup_group="USA_music_all_screens")
return rel_path, result
def save_usa_extra_html(extra_key, uploaded_file):
if extra_key not in USA_EXTRA_HTML_FILES:
return None
filename = preferred_main_upload_filename("USA_TV") if extra_key == "usa_trip_tv" else USA_EXTRA_HTML_FILES[extra_key]
target_path = os.path.join(usa_folder_for_extra_html_key(extra_key), filename)
result = atomic_save_upload(uploaded_file, target_path, backup_group="USA_TV_extra_html" if extra_key in USA_EXTRA_HTML_ROOTS else "USA_extra_html")
return filename, result
def save_file_to_multiple_targets(uploaded_file, base_folder, rel_paths, backup_group):
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
tmp_path = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"multi_upload_{stamp}_{os.getpid()}")
uploaded_file.save(tmp_path)
try:
size = os.path.getsize(tmp_path)
if size <= 0:
raise ValueError("uploaded file is empty")
sha = file_sha256(tmp_path)
results = []
for rel_path in rel_paths:
target_path = os.path.join(base_folder, rel_path)
ensure_parent_dir(target_path)
backup_path = backup_existing_file(target_path, backup_group)
tmp_target = target_path + ".tmp"
shutil.copy2(tmp_path, tmp_target)
os.replace(tmp_target, target_path)
results.append({"rel": rel_path, "path": target_path, "size": os.path.getsize(target_path), "sha256": file_sha256(target_path), "backup": backup_path})
return {"size": size, "sha256": sha, "targets": results}
finally:
try:
if os.path.exists(tmp_path):
os.remove(tmp_path)
except Exception:
pass
def save_japan_music(music_key, uploaded_file):
if music_key not in JAPAN_MUSIC_TARGETS:
return None
return save_audio_upload_to_multiple_targets(uploaded_file, FOLDERS["JAPAN"], JAPAN_MUSIC_TARGETS[music_key], "JAPAN_music")
def normalize_japan_image_rel(raw_name):
safe = safe_zip_relpath(raw_name)
if not safe:
return None
lower = safe.lower()
marker = "/media/images/"
if marker in lower:
idx = lower.index(marker) + len(marker)
return safe[idx:]
if lower.startswith("media/images/"):
return safe[len("media/images/"):]
if lower.startswith("images/"):
return safe[len("images/"):]
marker2 = "/images/"
if marker2 in lower:
idx = lower.index(marker2) + len(marker2)
return safe[idx:]
return None
def save_japan_images_zip(uploaded_file):
image_root = os.path.join(FOLDERS["JAPAN"], "media", "images")
os.makedirs(image_root, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
tmp_zip = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"japan_images_{stamp}.zip")
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
uploaded_file.save(tmp_zip)
imported, skipped, overwritten = [], [], []
try:
if os.path.getsize(tmp_zip) > JAPAN_IMAGES_ZIP_MAX_BYTES:
raise ValueError(f"zip_size_exceeded:{os.path.getsize(tmp_zip)}")
with zipfile.ZipFile(tmp_zip, "r") as zf:
infos = [info for info in zf.infolist() if not info.is_dir()]
if len(infos) > JAPAN_IMAGES_ZIP_MAX_FILES:
raise ValueError(f"too_many_files:{len(infos)}")
total_uncompressed = sum(max(0, info.file_size) for info in infos)
if total_uncompressed > JAPAN_IMAGES_ZIP_MAX_UNCOMPRESSED_BYTES:
raise ValueError(f"zip_uncompressed_size_exceeded:{total_uncompressed}")
for info in infos:
rel = normalize_japan_image_rel(info.filename)
base = os.path.basename((info.filename or "").replace("\\", "/"))
if not rel:
skipped.append(base + " — לא בתיקיית images/media/images")
continue
if info.file_size > JAPAN_IMAGES_ZIP_MAX_SINGLE_FILE_BYTES:
skipped.append(rel + " — גדול מדי")
continue
if file_ext(rel) not in JAPAN_IMAGE_EXTENSIONS:
skipped.append(rel + " — סיומת לא נתמכת")
continue
# Prevent traversal after normalization.
if not safe_zip_relpath(rel):
skipped.append(base + " — נתיב לא בטוח")
continue
target = os.path.join(image_root, *rel.split("/"))
ensure_parent_dir(target)
if os.path.exists(target):
overwritten.append(rel)
backup_existing_file(target, "JAPAN_images_zip")
tmp_target = target + ".tmp"
with zf.open(info, "r") as source, open(tmp_target, "wb") as dest:
shutil.copyfileobj(source, dest)
os.replace(tmp_target, target)
imported.append(rel)
finally:
try:
os.remove(tmp_zip)
except Exception:
pass
return sorted(imported), sorted(skipped), sorted(overwritten)
def normalize_trip_package_rel(trip, raw_name):
safe = safe_zip_relpath(raw_name)
if not safe:
return None
lower = safe.lower()
if trip == "USA":
if lower.startswith("usa/"):
return safe[4:]
# Also allow a package that contains USA files at root.
# Do not accept root index.html from Netlify ZIPs, because it is usually only a root redirect/landing file and would overwrite USA/index.html.
if lower in {"usa2027-presentation.html", "usa2027-presentation-tv.html", "usa2027-tv.html"} or lower.startswith("media/usa-presentation/"):
return safe
return None
if trip == "JAPAN":
if lower.startswith("japan/"):
return safe[6:]
if lower in {"index.html", "japan-experience.html", "japan-flight-intro.html", "japan-theme.mp3"} or lower.startswith("media/"):
return safe
return None
return None
def save_trip_package_zip(trip, uploaded_file):
if trip not in {"USA", "JAPAN"}:
return None
base_folder = FOLDERS[trip]
os.makedirs(base_folder, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
tmp_zip = os.path.join(ADMIN_UPLOAD_TMP_DIR, f"{trip.lower()}_package_{stamp}.zip")
os.makedirs(ADMIN_UPLOAD_TMP_DIR, exist_ok=True)
uploaded_file.save(tmp_zip)
imported, skipped, overwritten = [], [], []
try:
if os.path.getsize(tmp_zip) > TRIP_PACKAGE_ZIP_MAX_BYTES:
raise ValueError(f"zip_size_exceeded:{os.path.getsize(tmp_zip)}")
with zipfile.ZipFile(tmp_zip, "r") as zf:
infos = [info for info in zf.infolist() if not info.is_dir()]
if len(infos) > TRIP_PACKAGE_ZIP_MAX_FILES:
raise ValueError(f"too_many_files:{len(infos)}")
total_uncompressed = sum(max(0, info.file_size) for info in infos)
if total_uncompressed > TRIP_PACKAGE_ZIP_MAX_UNCOMPRESSED_BYTES:
raise ValueError(f"zip_uncompressed_size_exceeded:{total_uncompressed}")
for info in infos:
rel = normalize_trip_package_rel(trip, info.filename)
base = os.path.basename((info.filename or "").replace("\\", "/"))
if not rel:
skipped.append(base + " — לא שייך לחבילת " + trip)
continue
if info.file_size > TRIP_PACKAGE_ZIP_MAX_SINGLE_FILE_BYTES:
skipped.append(rel + " — גדול מדי")
continue
ext = file_ext(rel)
if ext not in TRIP_PACKAGE_EXTENSIONS:
skipped.append(rel + " — סיומת לא נתמכת")
continue
if not safe_zip_relpath(rel):
skipped.append(base + " — נתיב לא בטוח")
continue
target = os.path.join(base_folder, *rel.split("/"))
ensure_parent_dir(target)
if os.path.exists(target):
overwritten.append(rel)
backup_existing_file(target, f"{trip}_package_zip")
tmp_target = target + ".tmp"
with zf.open(info, "r") as source, open(tmp_target, "wb") as dest:
shutil.copyfileobj(source, dest)
os.replace(tmp_target, target)
imported.append(rel)
finally:
try:
os.remove(tmp_zip)
except Exception:
pass
return sorted(imported), sorted(skipped), sorted(overwritten)
def normalize_netlify_trip_key(trip):
trip = (trip or "").strip().upper()
if trip in {"USA", "US", "AMERICA"}:
return "USA"
if trip in {"JAPAN", "JP"}:
return "JAPAN"
return ""
def netlify_output_filename(trip):
trip = normalize_netlify_trip_key(trip)
stamp = time.strftime("%Y%m%d_%H%M%S")
if trip == "USA":
return f"USA2027_NETLIFY_DEPLOY_FROM_ADMIN_{stamp}.zip"
if trip == "JAPAN":
return f"JAPAN2026_NETLIFY_DEPLOY_FROM_ADMIN_{stamp}.zip"
return f"NETLIFY_DEPLOY_{stamp}.zip"
def netlify_is_static_file(path):
base = os.path.basename(path or "").lower()
if base in NETLIFY_SKIP_FILES:
return False
if base in NETLIFY_CONTROL_FILES:
return True
return file_ext(path) in NETLIFY_STATIC_EXTENSIONS
def netlify_should_copy_relpath(rel, profile=""):
rel = (rel or "").replace("\\", "/").lstrip("./")
low = rel.lower()
if profile == "USA_NETLIFY_PHONE":
if os.path.basename(low) in NETLIFY_USA_EXCLUDED_FILES:
return False
if any(low.startswith(prefix) for prefix in NETLIFY_USA_EXCLUDED_PREFIXES):
return False
return True
def netlify_copy_static_tree(source_root, target_root, prefix="", profile=""):
copied = []
if not os.path.isdir(source_root):
raise FileNotFoundError(source_root)
for dirpath, dirnames, filenames in os.walk(source_root):
dirnames[:] = [d for d in dirnames if d.lower() not in NETLIFY_SKIP_DIRS and not d.startswith(".")]
rel_dir = os.path.relpath(dirpath, source_root)
if rel_dir == ".":
rel_dir = ""
for name in filenames:
src = os.path.join(dirpath, name)
if not netlify_is_static_file(src):
continue
source_rel = os.path.join(rel_dir, name) if rel_dir else name
source_rel = source_rel.replace("\\", "/")
if not netlify_should_copy_relpath(source_rel, profile=profile):
continue
rel = source_rel
if prefix:
rel = prefix.strip("/") + "/" + rel
safe = safe_zip_relpath(rel)
if not safe:
continue
dst = os.path.join(target_root, *safe.split("/"))
ensure_parent_dir(dst)
shutil.copy2(src, dst)
copied.append(safe)
return sorted(copied, key=lambda x: x.lower())
def netlify_replace_attrs(html_text, replacer):
attr_re = re.compile(r"(?P<attr>\b(?:href|src|poster|data-href)\s*=\s*)(?P<quote>[\"'])(?P<url>.*?)(?P=quote)", re.I | re.S)
def repl(match):
url = match.group("url")
new_url = replacer(url)
if new_url is None:
return match.group(0)
return match.group("attr") + match.group("quote") + new_url + match.group("quote")
return attr_re.sub(repl, html_text or "")
def netlify_is_budget_url(url):
u = str(url or "").strip().lower()
return (
"beautiful-kitsune-dafd66.netlify.app" in u
or "moneyapp" in u
or "/moneyapp" in u
or "trip-budget" in u
or "vacation-budget" in u
)
def netlify_patch_trip_html(trip, html_text):
trip = normalize_netlify_trip_key(trip)
text = html_text or ""
def attr_replacer(url):
u = str(url or "").strip()
low = u.lower()
if netlify_is_budget_url(u):
return NETLIFY_BUDGET_APP_URL
if trip == "USA" and "usa2027-presentation" in low:
return "/USA/usa2027-presentation.html"
if trip == "JAPAN" and ("japan-flight-intro.html" in low or "japan-experience.html" in low):
return "./japan-flight-intro.html?from=trip"
return None
text = netlify_replace_attrs(text, attr_replacer)
# Also cover common JavaScript button patterns such as window.open('...').
if trip == "USA":
text = re.sub(r"https?://[^'\"\s<>]+/USA/usa2027-presentation\.html(?:\?[^'\"\s<>]*)?", "/USA/usa2027-presentation.html", text, flags=re.I)
text = re.sub(r"(?<![\w./-])(?:\./)?usa2027-presentation\.html(?:\?[^'\"\s<>]*)?", "/USA/usa2027-presentation.html", text, flags=re.I)
elif trip == "JAPAN":
text = re.sub(r"https?://[^'\"\s<>]+/JAPAN/japan-flight-intro\.html(?:\?[^'\"\s<>]*)?", "./japan-flight-intro.html?from=trip", text, flags=re.I)
text = re.sub(r"https?://[^'\"\s<>]+/JAPAN/japan-experience\.html(?:\?[^'\"\s<>]*)?", "./japan-flight-intro.html?from=trip", text, flags=re.I)
text = re.sub(r"https?://[^'\"\s<>]+/Moneyapp/?(?:index\.html)?(?:\?[^'\"\s<>]*)?", NETLIFY_BUDGET_APP_URL, text, flags=re.I)
text = re.sub(r"https?://beautiful-kitsune-dafd66\.netlify\.app(?:/index\.html)?/?", NETLIFY_BUDGET_APP_URL, text, flags=re.I)
return text
def netlify_patch_all_html_files(build_root, trip):
patched = []
trip = normalize_netlify_trip_key(trip)
for dirpath, _, filenames in os.walk(build_root):
for name in filenames:
if file_ext(name) not in {".html", ".htm"}:
continue
path = os.path.join(dirpath, name)
try:
with open(path, "r", encoding="utf-8", errors="replace") as fh:
text = fh.read()
original = text
# Remove private server links from generated Netlify bundles.
if trip == "USA":
text = re.sub(r"https?://work-room\.tail978ec2\.ts\.net/USA/", "/USA/", text, flags=re.I)
text = re.sub(r"https?://[^'\"\s<>]+/USA/", "/USA/", text, flags=re.I)
elif trip == "JAPAN":
text = re.sub(r"https?://work-room\.tail978ec2\.ts\.net/JAPAN/", "./", text, flags=re.I)
text = re.sub(r"https?://[^'\"\s<>]+/JAPAN/", "./", text, flags=re.I)
text = re.sub(r"https?://work-room\.tail978ec2\.ts\.net/Moneyapp/?(?:index\.html)?", NETLIFY_BUDGET_APP_URL, text, flags=re.I)
text = re.sub(r"https?://[^'\"\s<>]+/Moneyapp/?(?:index\.html)?", NETLIFY_BUDGET_APP_URL, text, flags=re.I)
if os.path.relpath(path, build_root).replace("\\", "/") in {"index.html", "USA/index.html"}:
text = netlify_patch_trip_html(trip, text)
if text != original:
with open(path, "w", encoding="utf-8", newline="") as fh:
fh.write(text)
patched.append(os.path.relpath(path, build_root).replace("\\", "/"))
except Exception:
raise
return sorted(patched, key=lambda x: x.lower())
def netlify_write_usa_root_index(build_root):
path = os.path.join(build_root, "index.html")
html = """<!DOCTYPE html>
<html lang=\"he\" dir=\"rtl\"><head><meta charset=\"utf-8\"><meta name=\"viewport\" content=\"width=device-width,initial-scale=1\"><title>טיול ארצות הברית 2027</title><meta http-equiv=\"refresh\" content=\"0; url=/USA/\"><script>location.replace('/USA/');</script></head><body style=\"font-family:Arial,sans-serif;direction:rtl;text-align:center;padding:40px\"><h1>טיול ארצות הברית 2027</h1><p>מעביר לקובץ הטיול...</p><p><a href=\"/USA/\">פתח טיול ארה״ב</a></p></body></html>"""
with open(path, "w", encoding="utf-8", newline="") as fh:
fh.write(html)
return "index.html"
def netlify_ensure_japan_top_anchor(build_root):
path = os.path.join(build_root, "index.html")
if not os.path.isfile(path):
return False
with open(path, "r", encoding="utf-8", errors="replace") as fh:
text = fh.read()
if 'id="top"' in text or "id='top'" in text:
return False
new_text = re.sub(r"(<body\b[^>]*>)", r"\1\n<div id=\"top\" style=\"position:absolute;top:0\"></div>", text, count=1, flags=re.I)
if new_text == text:
new_text = '<div id="top" style="position:absolute;top:0"></div>\n' + text
with open(path, "w", encoding="utf-8", newline="") as fh:
fh.write(new_text)
return True
def netlify_write_usa_redirects(build_root):
"""Ensure USA Netlify bundles have a root control file for the /USA entry path."""
path = os.path.join(build_root, "_redirects")
if os.path.isfile(path):
return False
redirects = "\n".join([
"/USA /USA/ 301",
"/USA/* /USA/:splat 200",
""
])
with open(path, "w", encoding="utf-8", newline="") as fh:
fh.write(redirects)
return True
def netlify_write_japan_redirects(build_root):
"""Ensure Japan Netlify bundles include the same redirect compatibility file used by the manual Netlify ZIPs."""
path = os.path.join(build_root, "_redirects")
if os.path.isfile(path):
return False
redirects = "\n".join([
"/JAPAN/japan-flight-intro.html /japan-flight-intro.html 200",
"/JAPAN/japan-experience.html /japan-experience.html 200",
"/JAPAN/ /index.html 200",
"/JAPAN/* /index.html 200",
"/japan-flight-intro.html /japan-flight-intro.html 200",
"/japan-experience.html /japan-experience.html 200",
""
])
with open(path, "w", encoding="utf-8", newline="") as fh:
fh.write(redirects)
return True
def netlify_local_ref_to_path(build_root, html_rel, raw_url):
url = str(raw_url or "").strip()
if not url or url.startswith("#") or url.startswith("?"):
return None
low = url.lower()
if low.startswith(("http://", "https://", "mailto:", "tel:", "sms:", "javascript:", "data:", "blob:", "about:", "//")):
return None
clean = url.split("#", 1)[0].split("?", 1)[0].strip()
if not clean:
return None
if clean.startswith("/"):
rel = clean.strip("/")
else:
base = os.path.dirname(html_rel)
rel = os.path.normpath(os.path.join(base, clean)).replace("\\", "/")
rel = rel.strip("/")
if not rel or rel == ".":
rel = "index.html"
return rel
def netlify_path_exists(build_root, rel):
safe = safe_zip_relpath(rel)
if not safe:
return False
path = os.path.join(build_root, *safe.split("/"))
if os.path.isfile(path):
return True
if os.path.isdir(path) and os.path.isfile(os.path.join(path, "index.html")):
return True
if safe.endswith("/") and os.path.isfile(os.path.join(build_root, *safe.split("/"), "index.html")):
return True
return False
def netlify_validate_build(build_root, trip):
trip = normalize_netlify_trip_key(trip)
errors, warnings, checked_refs = [], [], []
required = ["index.html"]
if trip == "USA":
required += ["USA/index.html", "USA/usa2027-presentation.html", "_redirects"]
elif trip == "JAPAN":
required += ["japan-flight-intro.html", "japan-experience.html", "_redirects"]
for rel in required:
if not netlify_path_exists(build_root, rel):
errors.append("חסר קובץ חובה: " + rel)
attr_re = re.compile(r"\b(?:href|src|poster|data-href)\s*=\s*[\"']([^\"']+)[\"']", re.I)
for dirpath, _, filenames in os.walk(build_root):
for name in filenames:
if file_ext(name) not in {".html", ".htm"}:
continue
path = os.path.join(dirpath, name)
html_rel = os.path.relpath(path, build_root).replace("\\", "/")
with open(path, "r", encoding="utf-8", errors="replace") as fh:
text = fh.read()
if "work-room.tail978ec2.ts.net" in text or "tail978ec2" in text:
errors.append("נשאר קישור לשרת הפרטי בתוך: " + html_rel)
for raw_url in attr_re.findall(text):
rel = netlify_local_ref_to_path(build_root, html_rel, raw_url)
if not rel:
continue
# Ignore non-file route style links that Netlify cannot validate statically, except known trip paths.
if not os.path.splitext(rel)[1] and not rel.endswith("/"):
candidate_dir = os.path.join(build_root, *rel.split("/"))
if not os.path.isdir(candidate_dir):
continue
checked_refs.append(html_rel + " -> " + rel)
if not netlify_path_exists(build_root, rel):
errors.append("קישור פנימי חסר: " + html_rel + " -> " + raw_url + " => " + rel)
if trip == "USA":
presentation_rel = "USA/usa2027-presentation.html"
presentation_path = os.path.join(build_root, "USA", "usa2027-presentation.html")
required_phone_assets = [
"USA/media/usa-presentation/images/budget.jpg",
"USA/media/usa-presentation/music/usa-roadtrip-theme.mp3",
"USA/media/usa-presentation/music/usa-part2-theme.mp3",
]
for rel in required_phone_assets:
if not netlify_path_exists(build_root, rel):
errors.append("חסר קובץ מצגת ארה״ב עדכני: " + rel)
if os.path.isfile(presentation_path):
with open(presentation_path, "r", encoding="utf-8", errors="replace") as fh:
presentation_text = fh.read()
normalized_text = presentation_text.replace("\\", "/")
if "media/usa-presentation/images-tv/" in normalized_text:
errors.append("מצגת הפלאפון מפנה בטעות לתיקיית images-tv: " + presentation_rel)
if "usa2027-presentation-tv.html" in normalized_text.lower():
errors.append("מצגת הפלאפון מכילה הפניה למצגת TV: " + presentation_rel)
budget_ref_re = re.compile(r"<section\b[^>]*data-key=['\"]budget['\"][\s\S]*?<img\b[^>]*src=['\"]([^'\"]+)['\"]", re.I)
budget_match = budget_ref_re.search(normalized_text)
if not budget_match:
errors.append("לא נמצא שקף תקציב עם תמונת budget במצגת ארה״ב")
else:
budget_src = budget_match.group(1).split("?", 1)[0].split("#", 1)[0]
if budget_src != "media/usa-presentation/images/budget.jpg":
errors.append("שקף התקציב אינו משתמש בתמונת התקציב המשותפת: " + budget_src)
budget_section_match = re.search(r'<section\b[^>]*data-key=[\'"]budget[\'"][^>]*>[\s\S]*?</section>', normalized_text, re.I)
if not budget_section_match:
errors.append("לא נמצא שקף התקציב המלא במצגת ארה״ב")
else:
budget_section = budget_section_match.group(0)
required_budget_texts = (
"תקציב — בלי דרמה",
"השקף הזה משאיר את המספרים בקובץ הטיול",
"המסגרת התקציבית מנוהלת בקובץ הטיול ובאפליקציית התקציב",
"את המספרים המדויקים משאירים למעקב התקציב",
)
for required_text in required_budget_texts:
if required_text not in budget_section:
errors.append("שקף התקציב של Netlify אינו גרסת ללא-סכומים; חסר מלל מאושר: " + required_text)
forbidden_budget_texts = ("$81,297", "₪243,890", "$51,430", "₪154,290", "$11,622", "$10,000")
for forbidden_text in forbidden_budget_texts:
if forbidden_text in budget_section:
errors.append("שקף התקציב של Netlify עדיין מכיל סכום אסור: " + forbidden_text)
if budget_section != USA_NETLIFY_BUDGET_SLIDE_HTML:
errors.append("שקף התקציב ב-ZIP אינו זהה לשקף ללא-הסכומים המאושר של Netlify")
private_presentation_path = os.path.join(FOLDERS["USA"], "usa2027-presentation.html")
if not os.path.isfile(private_presentation_path):
errors.append("מצגת השרת הפרטי חסרה ולכן לא ניתן לאמת סנכרון עם Netlify")
else:
with open(private_presentation_path, "r", encoding="utf-8", errors="replace") as fh:
private_presentation_text = fh.read()
dual_sync = _v57_validate_dual_presentation_sync(private_presentation_text, presentation_text)
if not dual_sync.get("ok"):
errors.extend(dual_sync.get("errors") or [])
# TV belongs to C:\TripServer\USA_TV and is deliberately excluded from the phone Netlify ZIP.
for forbidden in ("USA/usa2027-presentation-tv.html", "USA/usa2027-tv.html"):
if netlify_path_exists(build_root, forbidden):
errors.append("קובץ TV הוכנס בטעות לחבילת Netlify של הפלאפון: " + forbidden)
return {"ok": not errors, "errors": sorted(set(errors)), "warnings": sorted(set(warnings)), "checked_refs": len(checked_refs)}
def zip_directory_contents(source_dir, zip_path):
ensure_parent_dir(zip_path)
if os.path.exists(zip_path):
os.remove(zip_path)
files = []
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
for dirpath, _, filenames in os.walk(source_dir):
for name in sorted(filenames, key=lambda x: x.lower()):
full = os.path.join(dirpath, name)
rel = os.path.relpath(full, source_dir).replace("\\", "/")
safe = safe_zip_relpath(rel)
if not safe:
continue
zf.write(full, safe)
files.append(safe)
return files
def build_netlify_deploy_zip(trip):
trip = normalize_netlify_trip_key(trip)
if trip not in {"USA", "JAPAN"}:
raise ValueError("trip_not_supported")
source_root = FOLDERS[trip]
if not os.path.isdir(source_root):
raise FileNotFoundError(source_root)
stamp = time.strftime("%Y%m%d_%H%M%S")
os.makedirs(NETLIFY_BUILD_DIR, exist_ok=True)
os.makedirs(NETLIFY_EXPORT_DIR, exist_ok=True)
build_root = os.path.join(NETLIFY_BUILD_DIR, trip.lower() + "_" + stamp)
if os.path.exists(build_root):
shutil.rmtree(build_root)
os.makedirs(build_root, exist_ok=True)
copied = []
if trip == "USA":
copied.extend(netlify_copy_static_tree(source_root, build_root, prefix="USA", profile="USA_NETLIFY_PHONE"))
netlify_write_usa_root_index(build_root)
netlify_write_usa_redirects(build_root)
else:
copied.extend(netlify_copy_static_tree(source_root, build_root, prefix=""))
netlify_ensure_japan_top_anchor(build_root)
netlify_write_japan_redirects(build_root)
patched = netlify_patch_all_html_files(build_root, trip)
if trip == "USA":
presentation_path = os.path.join(build_root, "USA", "usa2027-presentation.html")
if not os.path.isfile(presentation_path):
raise RuntimeError("usa_netlify_presentation_missing")
with open(presentation_path, "r", encoding="utf-8", errors="strict") as fh:
presentation_html = fh.read()
# Netlify only: replace the ENTIRE budget slide with the approved no-amount
# slide. The shared budget.jpg is copied unchanged from the private server.
budget_slide_re = re.compile(
r'<section\b[^>]*class=[\'"][^\'"]*\bbudget\b[^\'"]*[\'"][^>]*data-key=[\'"]budget[\'"][^>]*>[\s\S]*?</section>',
re.I,
)
presentation_html, count = budget_slide_re.subn(
lambda _m: USA_NETLIFY_BUDGET_SLIDE_HTML,
presentation_html,
count=1,
)
if count != 1:
raise RuntimeError("usa_netlify_budget_slide_full_replace_failed")
with open(presentation_path, "w", encoding="utf-8", newline="") as fh:
fh.write(presentation_html)
patched.append("USA/usa2027-presentation.html#budget-slide-approved-no-amount")
validation = netlify_validate_build(build_root, trip)
if not validation.get("ok"):
raise RuntimeError("netlify_validation_failed:\n" + "\n".join(validation.get("errors") or []))
zip_path = os.path.join(NETLIFY_EXPORT_DIR, netlify_output_filename(trip))
files = zip_directory_contents(build_root, zip_path)
details = {
"trip": trip,
"source_root": source_root,
"build_root": build_root,
"zip_path": zip_path,
"files": len(files),
"copied": len(copied),
"patched_html": patched,
"validation": validation,
"budget_app_url": NETLIFY_BUDGET_APP_URL,
"presentation_link": "/USA/usa2027-presentation.html" if trip == "USA" else "./japan-flight-intro.html?from=trip",
"size_bytes": os.path.getsize(zip_path) if os.path.exists(zip_path) else 0,
}
return zip_path, details
def describe_path(path):
exists = os.path.exists(path)
item = {"path": path, "exists": exists}
if exists and os.path.isfile(path):
try:
item["size"] = os.path.getsize(path)
item["mtime"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(os.path.getmtime(path)))
item["sha256"] = file_sha256(path)[:16]
except Exception as exc:
item["error"] = repr(exc)
return item
def count_files(root, extensions=None):
total = 0
if not os.path.isdir(root):
return 0
for dirpath, _, filenames in os.walk(root):
for name in filenames:
if extensions is None or file_ext(name) in extensions:
total += 1
return total
def list_relative_files(root, extensions=None):
files = []
if not os.path.isdir(root):
return files
for dirpath, _, filenames in os.walk(root):
for name in filenames:
if extensions is None or file_ext(name) in extensions:
full = os.path.join(dirpath, name)
rel = os.path.relpath(full, root).replace("\\", "/")
files.append(rel)
return sorted(files, key=lambda x: x.lower())
def normalize_web_path(path):
value = str(path or "").strip().replace("\\", "/")
value = value.split("?", 1)[0].split("#", 1)[0]
while value.startswith("./"):
value = value[2:]
return value
def analyze_japan_presentation_images(presentation_path=None):
"""Count images that Japan experience actually uses, not just files in media/images."""
presentation_path = presentation_path or os.path.join(FOLDERS["JAPAN"], "japan-experience.html")
image_root = os.path.join(FOLDERS["JAPAN"], "media", "images")
folder_files = list_relative_files(image_root, JAPAN_IMAGE_EXTENSIONS)
result = {
"presentation_path": presentation_path,
"presentation_exists": os.path.isfile(presentation_path),
"folder_images": len(folder_files),
"image_slides": 0,
"external_image_uses": 0,
"embedded_image_uses": 0,
"remote_image_uses": 0,
"other_src_uses": 0,
"unique_external_images": 0,
"missing_external_images": 0,
"unused_folder_images": 0,
"chapter_slides": 0,
"notes": [],
"missing_external_list": [],
"unused_folder_list": [],
}
if not result["presentation_exists"]:
result["notes"].append("קובץ המצגת japan-experience.html לא נמצא, לכן נספרו רק קבצים בתיקיית התמונות.")
return result
try:
with open(presentation_path, "r", encoding="utf-8", errors="ignore") as fh:
html_text = fh.read()
except Exception as exc:
result["notes"].append("לא ניתן לקרוא את קובץ מצגת יפן: " + repr(exc))
return result
refs = [m.group(2).strip() for m in re.finditer(r'["\']src["\']\s*:\s*(["\'])(.*?)\1', html_text, re.S) if m.group(2).strip()]
if not refs:
refs = [m.group(2).strip() for m in re.finditer(r'\bsrc\s*:\s*(["\'])(.*?)\1', html_text, re.S) if m.group(2).strip()]
external = []
embedded = []
remote = []
other = []
for ref in refs:
clean = normalize_web_path(ref)
if clean.startswith("media/images/"):
external.append(clean[len("media/images/"):])
elif clean.startswith("data:image"):
embedded.append(ref)
elif clean.startswith("http://") or clean.startswith("https://"):
remote.append(ref)
else:
other.append(ref)
unique_external = sorted(set(external), key=lambda x: x.lower())
folder_lower = {f.lower(): f for f in folder_files}
used_lower = {f.lower() for f in unique_external}
missing = [rel for rel in unique_external if rel.lower() not in folder_lower]
unused = [rel for rel in folder_files if rel.lower() not in used_lower]
result.update({
"image_slides": len(refs),
"external_image_uses": len(external),
"embedded_image_uses": len(embedded),
"remote_image_uses": len(remote),
"other_src_uses": len(other),
"unique_external_images": len(unique_external),
"missing_external_images": len(missing),
"unused_folder_images": len(unused),
"chapter_slides": len(re.findall(r'["\']chapter["\']\s*:\s*true', html_text)),
"missing_external_list": missing[:40],
"unused_folder_list": unused[:40],
})
if result["image_slides"] == 0:
result["notes"].append("לא נמצאו src בשקפי מצגת יפן. ייתכן שמבנה המצגת השתנה.")
if result["missing_external_images"]:
result["notes"].append("יש תמונות שהמצגת מבקשת אבל אינן קיימות בתיקיית media/images.")
if result["unused_folder_images"]:
result["notes"].append("יש תמונות בתיקייה שאינן בשימוש במצגת הפעילה.")
return result
def compact_japan_usage_for_health():
usage = analyze_japan_presentation_images()
return {
"image_slides_total": usage.get("image_slides", 0),
"external_files_used": usage.get("external_image_uses", 0),
"embedded_images": usage.get("embedded_image_uses", 0),
"folder_images": usage.get("folder_images", 0),
"missing_external_images": usage.get("missing_external_images", 0),
"unused_folder_images": usage.get("unused_folder_images", 0),
"chapter_slides": usage.get("chapter_slides", 0),
"presentation_exists": usage.get("presentation_exists", False),
}
def render_admin_status():
rows = []
def add(label, path):
info = describe_path(path)
status = "✅ קיים" if info.get("exists") else "❌ חסר"
size = f"{info.get('size', 0):,}" if info.get("exists") and "size" in info else "—"
mtime = info.get("mtime", "—")
sha = info.get("sha256", "—")
rows.append(f"<tr><td>{html_lib.escape(label)}</td><td>{status}</td><td>{size}</td><td>{html_lib.escape(mtime)}</td><td>{html_lib.escape(sha)}</td><td><code>{html_lib.escape(path)}</code></td></tr>")
add("פאנל פעיל", ADMIN_SERVER_PATH)
add("טיול יפן ראשי", os.path.join(FOLDERS["JAPAN"], "index.html"))
add("פתיח טיסה יפן", os.path.join(FOLDERS["JAPAN"], "japan-flight-intro.html"))
add("מצגת יפן", os.path.join(FOLDERS["JAPAN"], "japan-experience.html"))
add("מוזיקת מצגת יפן", os.path.join(FOLDERS["JAPAN"], "media", "japan-experience", "music", "japan-theme.mp3"))
add("מוזיקת פתיח יפן", os.path.join(FOLDERS["JAPAN"], "media", "japan-flight", "music", "flight-intro.mp3"))
add("טיול ארה״ב ראשי", os.path.join(FOLDERS["USA"], "index.html"))
add("טיול ארה״ב TV", os.path.join(FOLDERS["USA_TV"], "usa2027-tv.html"))
add("מצגת ארה״ב", os.path.join(FOLDERS["USA"], "usa2027-presentation.html"))
add("מצגת ארה״ב TV", os.path.join(FOLDERS["USA_TV"], "usa2027-presentation-tv.html"))
add("שיר ארה״ב חלק 1 — פלאפון", os.path.join(FOLDERS["USA"], "media", "usa-presentation", "music", "usa-roadtrip-theme.mp3"))
add("שיר ארה״ב חלק 2 — פלאפון", os.path.join(FOLDERS["USA"], "media", "usa-presentation", "music", "usa-part2-theme.mp3"))
add("שיר ארה״ב חלק 1 — TV", os.path.join(FOLDERS["USA_TV"], "media", "usa-presentation", "music", "usa-roadtrip-theme.mp3"))
add("שיר ארה״ב חלק 2 — TV", os.path.join(FOLDERS["USA_TV"], "media", "usa-presentation", "music", "usa-part2-theme.mp3"))
add("תקציב טיולים", os.path.join(FOLDERS["Moneyapp"], "index.html"))
add("תקציב משפחתי", os.path.join(FOLDERS["MoneyHome"], "index.html"))
add("טופס פתיחת טיול", os.path.join(FOLDERS["TripForm"], "index.html"))
usa_img_count = count_files(os.path.join(FOLDERS["USA"], "media", "usa-presentation", "images"), USA_IMAGE_EXTENSIONS)
japan_usage = analyze_japan_presentation_images()
japan_img_count = japan_usage["folder_images"]
japan_usage_ok = japan_usage["presentation_exists"] and japan_usage["missing_external_images"] == 0
japan_usage_badge = "✅ תקין" if japan_usage_ok else "⚠️ לבדיקה"
japan_usage_details = (
f"{japan_usage['external_image_uses']} קבצים חיצוניים + "
f"{japan_usage['embedded_image_uses']} מוטמעות · "
f"בתיקייה: {japan_img_count} · "
f"לא בשימוש: {japan_usage['unused_folder_images']} · "
f"חסרות: {japan_usage['missing_external_images']}"
)
watchdog_info = watchdog_status(include_log=False)
watchdog_label = "ON" if watchdog_info.get("task_installed") else "OFF"
missing_html = ""
if japan_usage["missing_external_images"]:
missing_items = "".join(f"<li><code>{html_lib.escape(x)}</code></li>" for x in japan_usage["missing_external_list"][:20])
missing_html = f"<div class='warn'><b>תמונות חסרות למצגת יפן</b><ul>{missing_items}</ul></div>"
unused_html = ""
if japan_usage["unused_folder_images"]:
unused_items = "".join(f"<li><code>{html_lib.escape(x)}</code></li>" for x in japan_usage["unused_folder_list"][:20])
more = ""
if japan_usage["unused_folder_images"] > 20:
more = f"<p>ועוד {japan_usage['unused_folder_images'] - 20} קבצים שלא מוצגים כאן.</p>"
unused_html = f"<details class='details'><summary>תמונות בתיקיית יפן שלא בשימוש במצגת ({japan_usage['unused_folder_images']})</summary><ul>{unused_items}</ul>{more}</details>"
return f"""
<!DOCTYPE html>
<html lang="he" dir="rtl">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>סטטוס TripServer</title>
<style>
body{{margin:0;background:#0f172a;color:white;font-family:Arial;padding:24px}}.wrap{{max-width:1180px;margin:auto}}a{{color:#93c5fd}}table{{width:100%;border-collapse:collapse;background:#1e293b;border-radius:18px;overflow:hidden}}th,td{{border-bottom:1px solid #334155;padding:10px;text-align:right;vertical-align:top}}th{{background:#111827;color:#bfdbfe}}code{{direction:ltr;display:block;white-space:normal;color:#fde68a}}.cards{{display:grid;grid-template-columns:repeat(auto-fit,minmax(240px,1fr));gap:14px;margin:18px 0}}.card{{background:#1e293b;border:1px solid #334155;border-radius:18px;padding:16px}}.num{{font-size:30px;font-weight:900;color:#fde68a}}
.small{{color:#cbd5e1;font-size:13px;line-height:1.5;margin-top:6px}}.warn{{background:rgba(127,29,29,.38);border:1px solid rgba(248,113,113,.35);color:#fecaca;border-radius:16px;padding:14px;margin:14px 0}}.details{{background:#111827;border:1px solid #334155;border-radius:16px;padding:14px;margin:14px 0;color:#cbd5e1}}summary{{cursor:pointer;font-weight:900;color:#fde68a}}ul{{margin:10px 0 0}}
</style></head><body><div class="wrap"><a href="/admin">⬅ חזרה לפאנל</a><h1>סטטוס קבצים פעילים</h1><p>כאן רואים בפועל מה מותקן כרגע ב־C:\\TripServer, כולל גודל, תאריך שינוי ו־SHA קצר. זה נועד למנוע מצב של “העליתי אבל אני רואה גרסה ישנה”.</p><div class="cards"><div class="card"><div>קובצי מדיה מצגת ארה״ב</div><div class="num">{usa_img_count}</div><div class="small">כרטיסיות תמונה/סרטון מסונכרנות לשקפי ארה״ב.</div></div><div class="card"><div>מצגת יפן — תמונות בשימוש בפועל</div><div class="num">{japan_usage['image_slides']}</div><div class="small">{japan_usage_badge} · {japan_usage_details}</div></div><div class="card"><div>קבצים בתיקיית תמונות יפן</div><div class="num">{japan_img_count}</div><div class="small">זה מספר קבצים בתיקייה, לא מספר התמונות שהמצגת משתמשת בהן.</div></div><div class="card"><div>גרסת פאנל</div><div>{html_lib.escape(ADMIN_PANEL_VERSION)}</div></div></div>{missing_html}{unused_html}<table><thead><tr><th>רכיב</th><th>מצב</th><th>גודל bytes</th><th>עודכן</th><th>SHA16</th><th>נתיב</th></tr></thead><tbody>{''.join(rows)}</tbody></table></div></body></html>
"""
def validate_admin_server_upload(tmp_path):
size = os.path.getsize(tmp_path)
if size <= 0:
return False, "הקובץ ריק."
if size > ADMIN_MAX_UPLOAD_BYTES:
return False, f"הקובץ גדול מדי: {size:,} bytes. המגבלה היא {ADMIN_MAX_UPLOAD_BYTES:,} bytes."
try:
py_compile.compile(tmp_path, doraise=True)
except py_compile.PyCompileError as exc:
return False, "בדיקת קומפילציה נכשלה:\n" + str(exc)
except Exception as exc:
return False, "שגיאה בבדיקת הקובץ:\n" + repr(exc)
try:
with open(tmp_path, "r", encoding="utf-8", errors="ignore") as fh:
content = fh.read()
except Exception as exc:
return False, "לא ניתן לקרוא את הקובץ:\n" + repr(exc)
required_tokens = ["Flask", "@app.route", "app.run", "BASE_DIR", "ADMIN_PORT", "restart_process_later", "upload_admin_server", "run_self_test_suite", "build_watchdog_code", "watchdog_status"]
missing = [token for token in required_tokens if token not in content]
if missing:
return False, "הקובץ עבר קומפילציה, אבל לא נראה כמו admin_server.py של TripServer. חסרים סימנים: " + ", ".join(missing)
return True, "OK"
def extract_admin_server_from_zip_upload(zip_path, work_dir):
"""Validate a controlled admin update ZIP and extract only the admin Python file.
Supported package format:
- root admin_server.py, or
- exactly one root-level .py file that passes TripServer admin validation.
For safety, this route does not execute arbitrary scripts from the ZIP.
"""
size = os.path.getsize(zip_path)
if size <= 0:
return False, "קובץ ה־ZIP ריק.", None
if size > ADMIN_MAX_PACKAGE_UPLOAD_BYTES:
return False, f"קובץ ה־ZIP גדול מדי: {size:,} bytes. המגבלה היא {ADMIN_MAX_PACKAGE_UPLOAD_BYTES:,} bytes.", None
try:
with zipfile.ZipFile(zip_path, "r") as zf:
infos = zf.infolist()
if not infos:
return False, "קובץ ה־ZIP ריק.", None
unsafe = []
py_candidates = []
for info in infos:
name = (info.filename or "").replace("\\", "/")
clean = name.strip("/")
if not clean or info.is_dir():
continue
if clean.startswith("/") or ".." in clean.split("/"):
unsafe.append(name)
continue
# Keep package execution intentionally disabled; only extract admin Python.
if clean.lower() == "admin_server.py" or ("/" not in clean and clean.lower().endswith(".py")):
py_candidates.append(clean)
if unsafe:
return False, "קובץ ה־ZIP כולל נתיבים לא בטוחים: " + ", ".join(unsafe[:6]), None
if "admin_server.py" in py_candidates:
selected = "admin_server.py"
elif len(py_candidates) == 1:
selected = py_candidates[0]
else:
return False, "ZIP עדכון אדמין חייב לכלול admin_server.py בשורש, או קובץ Python יחיד בשורש החבילה.", None
os.makedirs(work_dir, exist_ok=True)
extracted_path = os.path.join(work_dir, "admin_server_from_zip.py")
with zf.open(selected, "r") as src, open(extracted_path, "wb") as dst:
shutil.copyfileobj(src, dst)
valid, details = validate_admin_server_upload(extracted_path)
if not valid:
try: os.remove(extracted_path)
except Exception: pass
return False, "admin_server.py שבתוך ה־ZIP נדחה:\n" + details, None
return True, f"ZIP תקין. נבחר קובץ: {selected}", extracted_path
except zipfile.BadZipFile:
return False, "קובץ ZIP לא תקין או פגום.", None
except Exception as exc:
return False, "שגיאה בבדיקת ZIP:\n" + repr(exc), None
def admin_status_page(title, message, ok=True, details=""):
color = "#16a34a" if ok else "#dc2626"
bg = "#052e1b" if ok else "#3f1d1d"
safe_title = html_lib.escape(title)
safe_message = html_lib.escape(message).replace("\n", "<br>")
safe_details = html_lib.escape(details).replace("\n", "<br>")
details_html = f"<pre>{safe_details}</pre>" if details else ""
return f"""
<!DOCTYPE html>
<html lang="he" dir="rtl">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{safe_title}</title>
<style>
body{{margin:0;font-family:Arial;background:#0f172a;color:white;min-height:100vh;display:flex;align-items:center;justify-content:center;padding:24px}}
.card{{width:min(780px,94vw);background:#1e293b;border-radius:24px;padding:26px;box-shadow:0 18px 70px rgba(0,0,0,.35)}}
.badge{{display:inline-block;background:{bg};border:1px solid {color};color:white;border-radius:999px;padding:8px 14px;font-weight:bold;margin-bottom:12px}}
h1{{margin:0 0 12px;font-size:30px}}
p{{line-height:1.7;color:#e2e8f0}}
pre{{white-space:pre-wrap;direction:ltr;text-align:left;background:#020617;color:#fde68a;border:1px solid #334155;border-radius:14px;padding:14px;overflow:auto;max-height:360px}}
a,button{{display:inline-block;margin:8px 0 0 8px;background:#2563eb;color:white;border:0;border-radius:12px;padding:12px 16px;text-decoration:none;font-weight:bold;cursor:pointer}}
button.restart{{background:#ea580c}}
.advanced-note{{margin-top:10px;color:#cbd5e1;font-size:13px;line-height:1.55}}
form{{display:inline}}
</style>
</head>
<body>
<div class="card">
<div class="badge">TripServer Admin</div>
<h1>{safe_title}</h1>
<p>{safe_message}</p>
{details_html}
<a href="/admin">חזרה לפאנל</a>
<form action="/restart_admin_server" method="post" onsubmit="return confirm('זו בדיקת אתחול מתקדמת. להפעיל מחדש את הפאנל עכשיו?')"><button class="restart">בדיקת אתחול פאנל — מתקדם</button></form>
<div class="advanced-note">מיועד לבדיקה יזומה בלבד אחרי עדכון/שחזור קובץ אדמין.</div>
</div>
</body>
</html>
"""
def backup_current_admin_server(prefix="admin_server_backup"):
os.makedirs(ADMIN_BACKUP_DIR, exist_ok=True)
stamp = time.strftime("%Y%m%d_%H%M%S")
backup_path = os.path.join(ADMIN_BACKUP_DIR, f"{prefix}_{stamp}.py")
shutil.copy2(ADMIN_SERVER_PATH, backup_path)
return backup_path
def latest_admin_backup():
os.makedirs(ADMIN_BACKUP_DIR, exist_ok=True)
backups = glob.glob(os.path.join(ADMIN_BACKUP_DIR, "admin_server_backup_*.py"))
if not backups:
return None
return max(backups, key=os.path.getmtime)
def log_control_bootstrap(message):
try:
os.makedirs(BASE_DIR, exist_ok=True)
with open(CONTROL_BOOTSTRAP_LOG_PATH, "a", encoding="utf-8") as fh:
fh.write(time.strftime("%Y-%m-%d %H:%M:%S") + " | control-bootstrap | " + str(message) + "\n")
except Exception:
pass
def quiet_pythonw_path():
exe = sys.executable or "python"
base = os.path.basename(exe).lower()
if base == "python.exe":
candidate = os.path.join(os.path.dirname(exe), "pythonw.exe")
if os.path.exists(candidate):
return candidate
return exe
def build_control_server_code():
return '# -*- coding: utf-8 -*-\nimport os, sys, time, json, shutil, subprocess, socket, threading, traceback\nfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer\nfrom urllib.parse import urlparse\nBASE_DIR = r"C:\\TripServer"\nADMIN_PORT = 8000\nCONTROL_PORT = 8010\nADMIN_SCRIPT = os.path.join(BASE_DIR, "admin_server.py")\nCONTROL_DIR = os.path.join(BASE_DIR, "control_server")\nPENDING_ADMIN = os.path.join(CONTROL_DIR, "pending_admin_server.py")\nREQUEST_JSON = os.path.join(CONTROL_DIR, "request.json")\nSTATUS_JSON = os.path.join(CONTROL_DIR, "status.json")\nFAILSAFE_STATUS_JSON = os.path.join(BASE_DIR, "admin_failsafe_status.json")\nLOG_PATH = os.path.join(CONTROL_DIR, "tripserver_control_server.log")\nBACKUP_DIR = os.path.join(BASE_DIR, "backups", "admin_server")\nPYTHON_EXE = sys.executable\nif os.path.basename(PYTHON_EXE).lower() == "pythonw.exe":\n c = os.path.join(os.path.dirname(PYTHON_EXE), "python.exe")\n if os.path.exists(c): PYTHON_EXE = c\nos.makedirs(CONTROL_DIR, exist_ok=True); os.makedirs(BACKUP_DIR, exist_ok=True)\n\ndef log(msg):\n try:\n with open(LOG_PATH, "a", encoding="utf-8") as f:\n f.write(time.strftime("%Y-%m-%d %H:%M:%S") + " | control-v34 | " + str(msg) + "\\n")\n except Exception: pass\n\ndef write_status(**kw):\n data={"updated_at":time.strftime("%Y-%m-%d %H:%M:%S"),"control_port":CONTROL_PORT,"admin_port":ADMIN_PORT,"python":PYTHON_EXE}\n data.update(kw)\n try:\n with open(STATUS_JSON,"w",encoding="utf-8") as f: json.dump(data,f,ensure_ascii=False,indent=2)\n except Exception as e: log("status write failed "+repr(e))\n return data\n\ndef port_open(port):\n s=socket.socket(socket.AF_INET,socket.SOCK_STREAM); s.settimeout(0.7)\n try: return s.connect_ex(("127.0.0.1",int(port)))==0\n finally:\n try: s.close()\n except Exception: pass\n\ndef admin_ping_ok():\n try:\n import urllib.request\n with urllib.request.urlopen("http://127.0.0.1:%s/ping.json"%ADMIN_PORT,timeout=1.5) as r: return r.status==200\n except Exception: return False\n\ndef pids_on_port(port):\n pids=[]\n if os.name!="nt": return pids\n try:\n cp=subprocess.run(["netstat","-ano","-p","tcp"],capture_output=True,text=True,timeout=8)\n for line in (cp.stdout or "").splitlines():\n low=line.lower()\n if (":"+str(port)) in low and "listening" in low:\n parts=line.split(); pid=parts[-1] if parts else ""\n if pid.isdigit() and int(pid) not in pids: pids.append(int(pid))\n except Exception as e: log("pids_on_port failed "+repr(e))\n return pids\n\ndef kill_admin():\n pids=pids_on_port(ADMIN_PORT); log("kill_admin pids="+repr(pids))\n for pid in pids:\n if pid==os.getpid(): continue\n try:\n cp=subprocess.run(["taskkill","/PID",str(pid),"/F"],capture_output=True,text=True,timeout=12)\n log("taskkill pid=%s rc=%s out=%s err=%s"%(pid,cp.returncode,(cp.stdout or "").strip(),(cp.stderr or "").strip()))\n except Exception as e: log("taskkill failed pid=%s %r"%(pid,e))\n deadline=time.time()+8\n while time.time()<deadline:\n if not port_open(ADMIN_PORT): return True\n time.sleep(0.25)\n return not port_open(ADMIN_PORT)\n\ndef start_admin():\n log("starting admin "+ADMIN_SCRIPT+" python="+PYTHON_EXE)\n flags=0\n if os.name=="nt": flags|=getattr(subprocess,"CREATE_NEW_PROCESS_GROUP",0)|getattr(subprocess,"CREATE_NO_WINDOW",0)|0x00000008\n out=open(os.path.join(BASE_DIR,"admin_stdout.log"),"a",encoding="utf-8",errors="replace")\n proc=subprocess.Popen([PYTHON_EXE,ADMIN_SCRIPT],cwd=BASE_DIR,stdout=out,stderr=out,stdin=subprocess.DEVNULL,creationflags=flags)\n log("started admin pid="+str(proc.pid)); write_status(last_start_pid=proc.pid,last_action="start_admin"); return proc.pid\n\ndef compile_ok(path):\n try:\n src=open(path,"r",encoding="utf-8",errors="replace").read(); compile(src,path,"exec"); return True,"OK"\n except Exception as e: return False,repr(e)\n\ndef sha256(path):\n import hashlib\n h=hashlib.sha256()\n with open(path,"rb") as f:\n for chunk in iter(lambda:f.read(1024*1024),b""): h.update(chunk)\n return h.hexdigest()\n\ndef write_failsafe_status(state,msg,**extra):\n try:\n data={"state":state,"message":msg,"updated_at":time.strftime("%Y-%m-%d %H:%M:%S"),"source":"control_server"}\n data.update(extra)\n tmp=FAILSAFE_STATUS_JSON+".tmp"\n with open(tmp,"w",encoding="utf-8") as f: json.dump(data,f,ensure_ascii=False,indent=2)\n os.replace(tmp,FAILSAFE_STATUS_JSON)\n except Exception as e: log("failsafe status write failed "+repr(e))\n\ndef no_pending_shutdown_msg(txt):\n low=(txt or "").lower()\n return ("1116" in low) or ("no shutdown was in progress" in low)\n\ndef cancel_pending_shutdown(reason="auto_cancel"):\n if os.name!="nt": return False,"not windows"\n try:\n cp=subprocess.run(["shutdown","/a"],capture_output=True,text=True,timeout=8)\n msg=(cp.stdout or cp.stderr or "").strip()\n log("shutdown /a reason=%s rc=%s out=%s err=%s"%(reason,cp.returncode,(cp.stdout or "").strip(),(cp.stderr or "").strip()))\n if cp.returncode==0:\n write_failsafe_status("auto_cancelled","Fail Safe בוטל אוטומטית אחרי שהאדמין עלה תקין.",windows_output=msg)\n return True,msg\n if no_pending_shutdown_msg(msg):\n write_failsafe_status("none","אין ריסטארט Fail Safe פעיל. Windows דיווח שלא היה כיבוי/ריסטארט בהמתנה.",windows_output=msg)\n return True,msg\n write_failsafe_status("cancel_failed","ניסיון ביטול אוטומטי החזיר שגיאה.",windows_output=msg,return_code=cp.returncode)\n return False,msg\n except Exception as e:\n log("shutdown /a failed "+repr(e)); write_failsafe_status("cancel_failed","שגיאה בביטול אוטומטי: "+repr(e)); return False,repr(e)\n\ndef apply_update(reason="manual"):\n log("apply_update begin reason="+str(reason))\n if not os.path.exists(PENDING_ADMIN):\n msg="pending admin not found"; log(msg); write_status(last_action="apply_update",ok=False,error=msg); return False,msg\n ok,info=compile_ok(PENDING_ADMIN)\n if not ok:\n log("pending compile failed "+info); write_status(last_action="apply_update",ok=False,error=info); return False,info\n new_sha=sha256(PENDING_ADMIN); stamp=time.strftime("%Y%m%d_%H%M%S"); backup=os.path.join(BACKUP_DIR,"admin_server_backup_"+stamp+".py")\n try:\n if os.path.exists(ADMIN_SCRIPT): shutil.copy2(ADMIN_SCRIPT,backup)\n except Exception as e: log("backup failed "+repr(e))\n kill_admin()\n try:\n os.replace(PENDING_ADMIN,ADMIN_SCRIPT); log("admin replaced sha="+new_sha+" backup="+backup)\n except Exception as e:\n log("replace failed "+repr(e)); write_status(last_action="apply_update",ok=False,error=repr(e)); return False,repr(e)\n pid=start_admin(); time.sleep(2); ping=admin_ping_ok(); cancel_ok=False; cancel_msg=""\n if ping: cancel_ok,cancel_msg=cancel_pending_shutdown("admin_update_ping_ok")\n write_status(last_action="apply_update",ok=True,new_sha=new_sha,backup=backup,started_pid=pid,ping_ok=ping,failsafe_cancel_ok=cancel_ok,failsafe_cancel_msg=cancel_msg); return True,"updated pid=%s ping=%s failsafe_cancel=%s"%(pid,ping,cancel_ok)\n\ndef restart_admin(reason="manual"):\n kill_admin(); pid=start_admin(); time.sleep(2); ping=admin_ping_ok(); write_status(last_action="restart_admin",ok=True,started_pid=pid,ping_ok=ping); return True,"started pid=%s ping=%s"%(pid,ping)\n\ndef process_request_file():\n if not os.path.exists(REQUEST_JSON): return\n try:\n with open(REQUEST_JSON,"r",encoding="utf-8") as f: req=json.load(f)\n except Exception as e: log("bad request json "+repr(e)); return\n try: not_before=float(req.get("not_before",0) or 0)\n except Exception: not_before=0\n if time.time()<not_before: return\n try: os.remove(REQUEST_JSON)\n except Exception: pass\n action=req.get("action")\n if action=="apply_update": apply_update(req.get("reason","request"))\n elif action=="restart_admin": restart_admin(req.get("reason","request"))\n else: log("unknown action "+repr(action))\n\ndef poll_loop():\n log("poll loop started")\n while True:\n try:\n process_request_file()\n if not port_open(ADMIN_PORT): log("admin port closed; starting admin"); start_admin()\n except Exception as e: log("poll error "+repr(e)+"\\n"+traceback.format_exc())\n time.sleep(2)\n\nclass Handler(BaseHTTPRequestHandler):\n server_version="TripServerControl/34"\n def _send(self,code,body,ctype="text/html; charset=utf-8"):\n if isinstance(body,str): body=body.encode("utf-8")\n self.send_response(code); self.send_header("Content-Type",ctype); self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body)\n def do_GET(self):\n path=urlparse(self.path).path\n if path=="/ping.json": self._send(200,json.dumps({"ok":True,"admin_ping":admin_ping_ok(),"admin_port_open":port_open(ADMIN_PORT)},ensure_ascii=False),"application/json; charset=utf-8"); return\n if path=="/status.json":\n data={}\n try:\n if os.path.exists(STATUS_JSON): data=json.load(open(STATUS_JSON,"r",encoding="utf-8"))\n except Exception as e: data={"error":repr(e)}\n data.update({"ok":True,"admin_ping":admin_ping_ok(),"admin_port_open":port_open(ADMIN_PORT),"pids":pids_on_port(ADMIN_PORT)})\n self._send(200,json.dumps(data,ensure_ascii=False,indent=2),"application/json; charset=utf-8"); return\n if path=="/log":\n try: txt=open(LOG_PATH,"r",encoding="utf-8",errors="replace").read()[-20000:]\n except Exception as e: txt=repr(e)\n self._send(200,"<pre style=\'white-space:pre-wrap;direction:ltr\'>"+txt.replace("<","<")+"</pre>"); return\n html="""<html><head><meta charset=\'utf-8\'><title>TripServer Control</title><style>body{font-family:Arial;background:#0b1220;color:#eef;direction:rtl;padding:30px}.card{max-width:760px;margin:auto;background:#111a2e;border:1px solid #334155;border-radius:22px;padding:24px}button{font-size:22px;padding:16px 22px;border-radius:16px;border:0;margin:8px;background:#2563eb;color:white}.danger{background:#dc2626}a{color:#93c5fd}</style></head><body><div class=\'card\'><h1>TripServer Control Server</h1><p>שרת בקרה חיצוני לפורט 8000. לא תלוי באדמין הראשי.</p><form method=\'post\' action=\'/restart\'><button>הפעל/אתחל אדמין ראשי</button></form><form method=\'post\' action=\'/apply-update\'><button class=\'danger\'>החל עדכון ממתין</button></form><p><a href=\'/status.json\'>Status JSON</a> · <a href=\'/log\'>Log</a></p></div></body></html>"""\n self._send(200,html)\n def do_POST(self):\n path=urlparse(self.path).path\n if path=="/restart": ok,msg=restart_admin("http"); self._send(200 if ok else 500,json.dumps({"ok":ok,"message":msg},ensure_ascii=False),"application/json; charset=utf-8"); return\n if path=="/apply-update": ok,msg=apply_update("http"); self._send(200 if ok else 500,json.dumps({"ok":ok,"message":msg},ensure_ascii=False),"application/json; charset=utf-8"); return\n self._send(404,"not found")\n def log_message(self,fmt,*args): log("http "+(fmt%args))\nif __name__=="__main__":\n log("starting control server python="+sys.executable)\n write_status(last_action="control_start",ok=True)\n threading.Thread(target=poll_loop,daemon=True).start()\n srv=ThreadingHTTPServer(("0.0.0.0",CONTROL_PORT),Handler)\n log("listening on port "+str(CONTROL_PORT))\n srv.serve_forever()\n'
def control_server_is_alive():
try:
with urllib.request.urlopen(f"http://127.0.0.1:{CONTROL_PORT}/ping.json", timeout=1.0) as resp:
return resp.status == 200
except Exception:
return False
def bootstrap_control_server(reason="startup"):
try:
os.makedirs(CONTROL_DIR, exist_ok=True)
code = build_control_server_code()
current = None
if os.path.exists(CONTROL_SCRIPT_PATH):
try:
current = open(CONTROL_SCRIPT_PATH, "r", encoding="utf-8", errors="replace").read()
except Exception:
current = None
if current != code:
compile(code, CONTROL_SCRIPT_PATH, "exec")
with open(CONTROL_SCRIPT_PATH, "w", encoding="utf-8") as fh:
fh.write(code)
log_control_bootstrap("control script written " + CONTROL_SCRIPT_PATH)
pyw = quiet_pythonw_path()
if os.name == "nt":
cp = subprocess.run(["schtasks", "/Create", "/TN", CONTROL_TASK_NAME, "/SC", "ONLOGON", "/TR", f'"{pyw}" "{CONTROL_SCRIPT_PATH}"', "/F"], capture_output=True, text=True, timeout=20)
log_control_bootstrap("schtasks create rc=%s out=%s err=%s" % (cp.returncode, (cp.stdout or "").strip(), (cp.stderr or "").strip()))
run = subprocess.run(["schtasks", "/Run", "/TN", CONTROL_TASK_NAME], capture_output=True, text=True, timeout=20)
log_control_bootstrap("schtasks run rc=%s out=%s err=%s" % (run.returncode, (run.stdout or '').strip(), (run.stderr or '').strip()))
if not control_server_is_alive():
flags = 0
if os.name == "nt":
flags |= getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | 0x00000008
subprocess.Popen([pyw, CONTROL_SCRIPT_PATH], cwd=BASE_DIR, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=flags)
log_control_bootstrap("direct control start requested via " + pyw)
return True, "OK"
except Exception as exc:
log_control_bootstrap("bootstrap failed " + repr(exc))
return False, repr(exc)
def write_control_request(action, reason="admin", pending_path=None):
os.makedirs(CONTROL_DIR, exist_ok=True)
req = {
"action": action,
"reason": reason,
"requested_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"not_before": time.time() + 2.5,
"admin_pid": os.getpid(),
"pending_admin_path": pending_path or CONTROL_PENDING_ADMIN_PATH,
}
with open(CONTROL_REQUEST_PATH, "w", encoding="utf-8") as fh:
json.dump(req, fh, ensure_ascii=False, indent=2)
return req
def log_admin_restart(message):
try:
os.makedirs(BASE_DIR, exist_ok=True)
with open(ADMIN_RESTART_LOG_PATH, "a", encoding="utf-8") as fh:
fh.write(time.strftime("%Y-%m-%d %H:%M:%S") + " | " + str(message) + "\n")
except Exception:
pass
def build_restart_helper_code(script_path, python_exe, work_dir, current_pid=None, backup_path=None, expected_sha=None, reason=""):
"""Build a simple and conservative restart helper.
v29 change: stop doing rollback/health gating inside the helper. The logs from
the real Windows machine showed the new Flask server starts, but the helper's
own health request sometimes times out and then incorrectly rolls back or
leaves the port in a bad state. Since uploaded admin files already pass
compile + TripServer validation before replacement, the safest behavior is:
wait, release port 8000, start admin_server.py, write logs, exit. Watchdog and
the browser then verify availability through /ping.json.
"""
return "\n".join([
"# -*- coding: utf-8 -*-",
"import os",
"import sys",
"import subprocess",
"import time",
"import socket",
"import hashlib",
"",
f"script_path = {script_path!r}",
f"python_exe = {python_exe!r}",
f"work_dir = {work_dir!r}",
f"log_path = {ADMIN_RESTART_LOG_PATH!r}",
f"admin_port = {ADMIN_PORT!r}",
f"current_pid = {int(current_pid or 0)!r}",
f"backup_path = {backup_path!r}",
f"expected_sha = {expected_sha!r}",
f"reason = {reason!r}",
"",
"def log(message):",
" try:",
" os.makedirs(os.path.dirname(log_path), exist_ok=True)",
" with open(log_path, 'a', encoding='utf-8') as fh:",
" fh.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | restart-v29 | ' + str(message) + '\\n')",
" except Exception:",
" pass",
"",
"def file_sha256(path):",
" h = hashlib.sha256()",
" with open(path, 'rb') as fh:",
" for chunk in iter(lambda: fh.read(1024 * 1024), b''):",
" h.update(chunk)",
" return h.hexdigest()",
"",
"def port_open():",
" try:",
" with socket.create_connection(('127.0.0.1', int(admin_port)), timeout=0.7):",
" return True",
" except Exception:",
" return False",
"",
"def wait_port_closed(timeout=12):",
" deadline = time.time() + timeout",
" while time.time() < deadline:",
" if not port_open():",
" return True",
" time.sleep(0.4)",
" return not port_open()",
"",
"def pids_on_port():",
" pids = []",
" if os.name != 'nt':",
" return pids",
" try:",
" cp = subprocess.run(['netstat', '-ano', '-p', 'tcp'], capture_output=True, text=True, shell=False, timeout=10)",
" out = (cp.stdout or '') + '\\n' + (cp.stderr or '')",
" marker = ':' + str(admin_port)",
" for line in out.splitlines():",
" if marker not in line or 'listen' not in line.lower():",
" continue",
" parts = line.split()",
" if parts:",
" try:",
" pid = int(parts[-1])",
" if pid and pid not in pids:",
" pids.append(pid)",
" except Exception:",
" pass",
" except Exception as exc:",
" log('netstat failed: ' + repr(exc))",
" return pids",
"",
"def kill_pid(pid):",
" try:",
" pid = int(pid)",
" except Exception:",
" return False",
" if pid <= 0 or pid == os.getpid():",
" log('skip kill pid=' + str(pid) + ' self=' + str(os.getpid()))",
" return False",
" try:",
" if os.name == 'nt':",
" # No /T. Kill only the process that owns port 8000, never a tree.",
" cp = subprocess.run(['taskkill', '/PID', str(pid), '/F'], capture_output=True, text=True, shell=False, timeout=12)",
" log('taskkill pid=' + str(pid) + ' rc=' + str(cp.returncode) + ' out=' + ((cp.stdout or cp.stderr or "").strip()[:500]))",
" return cp.returncode == 0",
" os.kill(pid, 9)",
" return True",
" except Exception as exc:",
" log('kill failed pid=' + str(pid) + ' err=' + repr(exc))",
" return False",
"",
"def release_port():",
" # Give the old request time to finish and Flask to exit if it exits naturally.",
" if wait_port_closed(10):",
" log('port closed naturally')",
" return",
" pids = pids_on_port()",
" log('port still open; killing owners without rollback; current_pid=' + str(current_pid) + ' pids=' + str(pids))",
" for pid in pids:",
" kill_pid(pid)",
" wait_port_closed(10)",
" log('port open after release=' + str(port_open()) + ' pids=' + str(pids_on_port()))",
"",
"def start_server():",
" flags = 0",
" if os.name == 'nt':",
" flags = getattr(subprocess, 'CREATE_NEW_PROCESS_GROUP', 0) | getattr(subprocess, 'DETACHED_PROCESS', 0) | getattr(subprocess, 'CREATE_BREAKAWAY_FROM_JOB', 0)",
" env = os.environ.copy()",
" env.setdefault('PYTHONIOENCODING', 'utf-8')",
" os.makedirs(os.path.dirname(log_path), exist_ok=True)",
" with open(log_path, 'a', encoding='utf-8') as log_file:",
" log_file.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | restart-v29 | starting admin: ' + script_path + '\\n')",
" log_file.flush()",
" proc = subprocess.Popen([python_exe, script_path], cwd=work_dir, stdin=subprocess.DEVNULL, stdout=log_file, stderr=log_file, close_fds=(False if os.name == 'nt' else True), creationflags=flags, start_new_session=(os.name != 'nt'), env=env)",
" log_file.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | restart-v29 | started admin pid=' + str(proc.pid) + '\\n')",
" log_file.flush()",
" return proc.pid",
"",
"def main():",
" log('simple restart started reason=' + str(reason) + ' script=' + script_path + ' python=' + python_exe + ' expected_sha=' + str(expected_sha) + ' backup=' + str(backup_path))",
" time.sleep(2.5)",
" if expected_sha:",
" try:",
" actual = file_sha256(script_path)",
" log('active sha=' + actual + ' expected=' + str(expected_sha))",
" except Exception as exc:",
" log('sha check failed: ' + repr(exc))",
" release_port()",
" start_server()",
" # Do not call /health or /ping here and do not roll back. Previous versions",
" # produced false timeout failures even though Flask was already running.",
" log('simple restart helper finished; browser/watchdog will verify /ping.json')",
" return 0",
"",
"if __name__ == '__main__':",
" raise SystemExit(main())",
"",
]) + "\n"
def validate_restart_helper_source(helper_code):
try:
compile(helper_code, "<tripserver_restart_helper>", "exec")
return True, "OK"
except Exception as exc:
return False, repr(exc)
def write_restart_helper(script_path, python_exe, work_dir, backup_path=None, expected_sha=None, reason=""):
helper_code = build_restart_helper_code(script_path, python_exe, work_dir, os.getpid(), backup_path, expected_sha, reason)
ok, info = validate_restart_helper_source(helper_code)
if not ok:
raise RuntimeError("Generated restart helper is invalid: " + info)
ensure_parent_dir(ADMIN_RESTART_HELPER_PATH)
with open(ADMIN_RESTART_HELPER_PATH, "w", encoding="utf-8") as fh:
fh.write(helper_code)
py_compile.compile(ADMIN_RESTART_HELPER_PATH, doraise=True)
return ADMIN_RESTART_HELPER_PATH
def restart_process_later(backup_path=None, expected_sha=None, reason="manual_restart"):
"""Hard self-restart for Windows/iPhone admin workflow.
v23: the helper can force-release a stuck port safely and roll back to the previous admin
backup if the newly uploaded admin does not pass /ping.json after boot.
"""
time.sleep(1.2)
script_path = os.path.abspath(ADMIN_SERVER_PATH)
python_exe = sys.executable or "python"
work_dir = os.path.dirname(script_path) or BASE_DIR
try:
log_admin_restart(f"hard restart requested | reason={reason} | version={ADMIN_PANEL_VERSION} | script={script_path} | python={python_exe} | pid={os.getpid()} | backup={backup_path} | expected_sha={expected_sha} | argv={sys.argv}")
helper_path = write_restart_helper(script_path, python_exe, work_dir, backup_path=backup_path, expected_sha=expected_sha, reason=reason)
flags = 0
if os.name == "nt":
flags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0)
subprocess.Popen(
[python_exe, helper_path],
cwd=work_dir,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=(False if os.name == "nt" else True),
creationflags=flags,
start_new_session=(os.name != "nt"),
)
log_admin_restart("hard restart helper launched: " + helper_path)
os._exit(0)
except Exception as exc:
log_admin_restart("hard restart helper failed: " + repr(exc))
try:
subprocess.Popen([python_exe, script_path], cwd=work_dir, close_fds=True, start_new_session=(os.name != "nt"))
os._exit(0)
except Exception as exc2:
log_admin_restart("direct restart fallback failed: " + repr(exc2))
try:
os.execv(python_exe, [python_exe, script_path])
except Exception as exc3:
log_admin_restart("execv fallback failed: " + repr(exc3))
os._exit(1)
def launch_restart_helper_via_scheduler(helper_path, python_exe, work_dir):
"""Launch restart helper as an independent Windows Scheduled Task.
This is more reliable than subprocess.Popen from the Flask process because the
helper is not a child of the admin process it is about to kill. On Windows,
self-restart helpers launched as child processes can be terminated together
with the old server or stuck in the same job/process group.
"""
if os.name != "nt":
raise RuntimeError("Task Scheduler restart is supported only on Windows")
quiet_python = get_pythonw_executable(python_exe)
task_command = f'"{quiet_python}" "{helper_path}"'
# Create a normal on-demand task. The time is deliberately far in the future;
# we immediately run it with /Run, and then overwrite it on the next update.
create_cmd = [
"schtasks", "/Create",
"/TN", RESTART_TASK_NAME,
"/TR", task_command,
"/SC", "ONCE",
"/ST", "23:59",
"/F",
]
cp_create = subprocess.run(create_cmd, capture_output=True, text=True, shell=False, timeout=20)
create_out = (cp_create.stdout or cp_create.stderr or "").strip()
if cp_create.returncode != 0:
raise RuntimeError("Failed to create restart task: " + (create_out or str(cp_create.returncode)))
cp_run = subprocess.run(["schtasks", "/Run", "/TN", RESTART_TASK_NAME], capture_output=True, text=True, shell=False, timeout=20)
run_out = (cp_run.stdout or cp_run.stderr or "").strip()
if cp_run.returncode != 0:
raise RuntimeError("Failed to run restart task: " + (run_out or str(cp_run.returncode)))
log_admin_restart("restart helper launched via Task Scheduler | task=" + RESTART_TASK_NAME + " | python=" + quiet_python + " | create=" + create_out + " | run=" + run_out)
return {"task": RESTART_TASK_NAME, "python": quiet_python, "create": create_out, "run": run_out}
def launch_safe_external_restart(backup_path=None, expected_sha=None, reason="safe_external_restart"):
"""Launch a detached restart helper through Task Scheduler.
v27: this is the important architectural fix. The helper is no longer a
direct child of the Flask/admin process. It is started by Windows Task
Scheduler, then it can safely kill the old process on port 8000 and start the
new admin without dying with its parent.
"""
script_path = os.path.abspath(ADMIN_SERVER_PATH)
python_exe = sys.executable or "python"
work_dir = os.path.dirname(script_path) or BASE_DIR
log_admin_restart(
f"scheduler restart requested | reason={reason} | version={ADMIN_PANEL_VERSION} | script={script_path} | python={python_exe} | pid={os.getpid()} | backup={backup_path} | expected_sha={expected_sha}"
)
helper_path = write_restart_helper(script_path, python_exe, work_dir, backup_path=backup_path, expected_sha=expected_sha, reason=reason)
try:
task_info = launch_restart_helper_via_scheduler(helper_path, python_exe, work_dir)
return helper_path + " | task=" + str(task_info.get("task")) + " | python=" + str(task_info.get("python"))
except Exception as exc:
# Fallback only: use direct detached Popen if Task Scheduler is unavailable.
# This preserves some functionality, but the scheduled task path is the
# intended reliable path on the user's Windows server.
log_admin_restart("Task Scheduler restart launch failed, falling back to detached Popen: " + repr(exc))
flags = 0
if os.name == "nt":
flags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0)
quiet_python = get_pythonw_executable(python_exe)
subprocess.Popen(
[quiet_python, helper_path],
cwd=work_dir,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=(False if os.name == "nt" else True),
creationflags=flags,
start_new_session=(os.name != "nt"),
)
log_admin_restart("restart helper launched through fallback detached Popen: " + helper_path)
return helper_path + " | fallback=detached-popen"
def build_watchdog_code(script_path, python_exe, work_dir):
"""Build a one-shot watchdog script for Windows Task Scheduler.
v28 change: if the port is open but /ping.json fails repeatedly, the watchdog
treats it as a stuck admin process, kills the process that owns port 8000, and
starts admin_server.py again. This is the case that forced a physical reboot.
"""
return "\n".join([
"# -*- coding: utf-8 -*-",
"import os",
"import sys",
"import time",
"import socket",
"import subprocess",
"import urllib.request",
"import json",
"",
f"script_path = {script_path!r}",
f"python_exe = {python_exe!r}",
f"work_dir = {work_dir!r}",
f"admin_port = {ADMIN_PORT!r}",
f"health_url = {'http://127.0.0.1:%s/ping.json' % ADMIN_PORT!r}",
f"log_path = {WATCHDOG_LOG_PATH!r}",
f"state_path = {WATCHDOG_STATE_PATH!r}",
f"cooldown_seconds = {WATCHDOG_MIN_RESTART_INTERVAL_SECONDS!r}",
"",
"def log(message):",
" try:",
" os.makedirs(os.path.dirname(log_path), exist_ok=True)",
" with open(log_path, 'a', encoding='utf-8') as fh:",
" fh.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | watchdog-v28 | ' + str(message) + '\\n')",
" except Exception:",
" pass",
"",
"def load_state():",
" try:",
" raw = open(state_path, 'r', encoding='utf-8').read().strip()",
" if not raw:",
" return {}",
" try:",
" data = json.loads(raw)",
" return data if isinstance(data, dict) else {}",
" except Exception:",
" return {'last_start': float(raw)}",
" except Exception:",
" return {}",
"",
"def save_state(state):",
" try:",
" os.makedirs(os.path.dirname(state_path), exist_ok=True)",
" with open(state_path, 'w', encoding='utf-8') as fh:",
" json.dump(state, fh)",
" except Exception:",
" pass",
"",
"def port_open():",
" try:",
" with socket.create_connection(('127.0.0.1', int(admin_port)), timeout=2.0):",
" return True",
" except Exception:",
" return False",
"",
"def pids_on_port():",
" pids = []",
" if os.name != 'nt':",
" return pids",
" try:",
" cp = subprocess.run(['netstat', '-ano', '-p', 'tcp'], capture_output=True, text=True, shell=False, timeout=10)",
" out = (cp.stdout or '') + '\\n' + (cp.stderr or '')",
" marker = ':' + str(admin_port)",
" for line in out.splitlines():",
" if marker not in line or 'listen' not in line.lower():",
" continue",
" parts = line.split()",
" if parts:",
" try:",
" pid = int(parts[-1])",
" if pid and pid not in pids:",
" pids.append(pid)",
" except Exception:",
" pass",
" except Exception as exc:",
" log('netstat failed: ' + repr(exc))",
" return pids",
"",
"def kill_port_owner(pid):",
" # v23: avoid /T even in watchdog. If run-once was launched from the admin",
" # panel, /T could kill the watchdog child together with the stuck admin.",
" try:",
" pid = int(pid)",
" except Exception:",
" return False",
" if pid <= 0 or pid == os.getpid():",
" log('skip killing pid=' + str(pid) + ' self=' + str(os.getpid()))",
" return False",
" try:",
" if os.name == 'nt':",
" cp = subprocess.run(['taskkill', '/PID', str(pid), '/F'], capture_output=True, text=True, shell=False, timeout=12)",
" log('taskkill-port-owner pid=' + str(pid) + ' rc=' + str(cp.returncode) + ' out=' + ((cp.stdout or cp.stderr or "").strip()[:500]))",
" return cp.returncode == 0",
" os.kill(pid, 9)",
" return True",
" except Exception as exc:",
" log('kill port owner failed pid=' + str(pid) + ' err=' + repr(exc))",
" return False",
"",
"def wait_port_closed(timeout=12):",
" deadline = time.time() + timeout",
" while time.time() < deadline:",
" if not port_open():",
" return True",
" time.sleep(0.6)",
" return not port_open()",
"",
"def health_ok():",
" try:",
" req = urllib.request.Request(health_url, headers={'Accept': 'application/json'})",
" with urllib.request.urlopen(req, timeout=4.0) as resp:",
" raw = resp.read(8192).decode('utf-8', errors='replace')",
" data = json.loads(raw)",
" return bool(data.get('ok')), 'OK' if data.get('ok') else raw[:500]",
" except Exception as exc:",
" return False, repr(exc)",
"",
"def recently_started(state):",
" try:",
" last = float(state.get('last_start') or 0)",
" return (time.time() - last) < float(cooldown_seconds)",
" except Exception:",
" return False",
"",
"def mark_started(state):",
" state['last_start'] = time.time()",
" state['failure_count'] = 0",
" save_state(state)",
"",
"def script_compiles():",
" try:",
" import py_compile",
" py_compile.compile(script_path, doraise=True)",
" return True, 'OK'",
" except Exception as exc:",
" return False, repr(exc)",
"",
"def launch_admin(state):",
" flags = 0",
" if os.name == 'nt':",
" flags = getattr(subprocess, 'CREATE_NEW_PROCESS_GROUP', 0) | getattr(subprocess, 'DETACHED_PROCESS', 0)",
" env = os.environ.copy()",
" env.setdefault('PYTHONIOENCODING', 'utf-8')",
" os.makedirs(os.path.dirname(log_path), exist_ok=True)",
" with open(log_path, 'a', encoding='utf-8') as log_file:",
" log_file.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | watchdog-v28 | starting admin: ' + script_path + '\\n')",
" log_file.flush()",
" proc = subprocess.Popen(",
" [python_exe, script_path],",
" cwd=work_dir,",
" stdin=subprocess.DEVNULL,",
" stdout=log_file,",
" stderr=log_file,",
" close_fds=(False if os.name == 'nt' else True),",
" creationflags=flags,",
" start_new_session=(os.name != 'nt'),",
" env=env,",
" )",
" log_file.write(time.strftime('%Y-%m-%d %H:%M:%S') + ' | watchdog-v28 | started pid=' + str(proc.pid) + '\\n')",
" mark_started(state)",
"",
"def main():",
" state = load_state()",
" ok, info = health_ok()",
" if ok:",
" if state.get('failure_count'):",
" state['failure_count'] = 0",
" save_state(state)",
" return 0",
" fail_count = int(state.get('failure_count') or 0) + 1",
" state['failure_count'] = fail_count",
" save_state(state)",
" is_port_open = port_open()",
" log('health failed count=' + str(fail_count) + ' port_open=' + str(is_port_open) + ' info=' + str(info))",
" if recently_started(state):",
" log('restart cooldown active; skip this minute')",
" return 3",
" if is_port_open:",
" pids = pids_on_port()",
" if fail_count < 2:",
" log('port is open but health failed; waiting for second failure before hard kill. pids=' + str(pids))",
" return 2",
" log('port open + repeated health failure; killing stuck pids=' + str(pids))",
" for pid in pids:",
" kill_port_owner(pid)",
" wait_port_closed(14)",
" if not os.path.exists(script_path):",
" log('admin script not found: ' + script_path)",
" return 4",
" ok_compile, compile_info = script_compiles()",
" if not ok_compile:",
" log('admin script does not compile; not starting. info=' + str(compile_info))",
" return 5",
" try:",
" launch_admin(state)",
" return 10",
" except Exception as exc:",
" log('failed to launch admin: ' + repr(exc))",
" return 6",
"",
"if __name__ == '__main__':",
" raise SystemExit(main())",
"",
]) + "\n"
def validate_watchdog_source(watchdog_code):
try:
compile(watchdog_code, "<tripserver_watchdog>", "exec")
return True, "OK"
except Exception as exc:
return False, repr(exc)
def write_watchdog_script():
script_path = os.path.abspath(ADMIN_SERVER_PATH)
python_exe = sys.executable or "python"
work_dir = os.path.dirname(script_path) or BASE_DIR
source = build_watchdog_code(script_path, python_exe, work_dir)
ok, info = validate_watchdog_source(source)
if not ok:
raise RuntimeError("Generated watchdog script is invalid: " + info)
ensure_parent_dir(WATCHDOG_SCRIPT_PATH)
with open(WATCHDOG_SCRIPT_PATH, "w", encoding="utf-8") as fh:
fh.write(source)
py_compile.compile(WATCHDOG_SCRIPT_PATH, doraise=True)
return WATCHDOG_SCRIPT_PATH
def read_log_tail(path, max_lines=18):
try:
with open(path, "r", encoding="utf-8", errors="replace") as fh:
lines = fh.readlines()
return "".join(lines[-max_lines:]).strip()
except Exception:
return ""
def safe_int(value, default, min_value=None, max_value=None):
try:
parsed = int(value)
except Exception:
parsed = default
if min_value is not None:
parsed = max(min_value, parsed)
if max_value is not None:
parsed = min(max_value, parsed)
return parsed
def log_file_summary(log_key, max_lines=LOG_TAIL_DEFAULT_LINES):
meta = LOG_FILE_REGISTRY.get(log_key)
if not meta:
return None
path = meta["path"]
exists = os.path.exists(path)
size = os.path.getsize(path) if exists else 0
updated = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(os.path.getmtime(path))) if exists else "לא קיים עדיין"
tail = read_log_tail(path, max_lines=max_lines) if exists else "אין עדיין לוג. הוא יווצר אחרי פעולה ראשונה."
return {
"key": log_key,
"label": meta["label"],
"note": meta.get("note", ""),
"path": path,
"exists": exists,
"size": size,
"size_label": format_bytes(size),
"updated": updated,
"tail": tail,
}
def build_logs_payload(max_lines=LOG_TAIL_DEFAULT_LINES):
max_lines = safe_int(max_lines, LOG_TAIL_DEFAULT_LINES, 20, LOG_TAIL_MAX_LINES)
items = []
for key in LOG_FILE_REGISTRY:
info = log_file_summary(key, max_lines=max_lines)
if info:
items.append(info)
return {
"ok": True,
"version": ADMIN_PANEL_VERSION,
"checked_at": time.strftime("%Y-%m-%d %H:%M:%S"),
"lines": max_lines,
"logs": items,
}
def render_admin_logs():
payload = build_logs_payload(request.args.get("lines", LOG_TAIL_DEFAULT_LINES))
cards = []
for info in payload["logs"]:
badge_cls = "ok" if info["exists"] else "warn"
badge_text = "קיים" if info["exists"] else "לא נוצר עדיין"
tail = html_lib.escape(info["tail"] or "אין תוכן להצגה.").replace("\n", "<br>")
cards.append(f'''
<div class="log-card">
<div class="log-head">
<div><h2>{html_lib.escape(info["label"])}</h2><p>{html_lib.escape(info["note"])}</p></div>
<span class="badge {badge_cls}">{badge_text}</span>
</div>
<div class="meta"><b>נתיב</b><code>{html_lib.escape(info["path"])}</code></div>
<div class="stats"><span>גודל: <b>{html_lib.escape(info["size_label"])}</b></span><span>עודכן: <b>{html_lib.escape(info["updated"])}</b></span></div>
<pre>{tail}</pre>
</div>''')
return f'''
<!DOCTYPE html>
<html lang="he" dir="rtl"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>לוגים TripServer</title>
<style>
body{{margin:0;background:#07111f;color:white;font-family:Arial;padding:22px}}.wrap{{max-width:1120px;margin:auto}}a{{color:#dbeafe;text-decoration:none}}.hero{{background:linear-gradient(135deg,#1e3a8a,#0f172a 62%,#111827);border:1px solid #1d4ed8;border-radius:28px;padding:22px;margin-bottom:16px;box-shadow:0 20px 70px rgba(0,0,0,.28)}}h1{{margin:0 0 8px;font-size:34px}}p{{color:#cbd5e1;line-height:1.6}}.actions{{display:flex;gap:10px;flex-wrap:wrap;margin-top:14px}}.actions a{{background:#2563eb;color:white;border-radius:999px;padding:10px 14px;font-weight:900}}.actions a.secondary{{background:#1e293b;border:1px solid #334155}}.log-card{{background:#101c31;border:1px solid #253650;border-radius:24px;padding:18px;margin:14px 0;box-shadow:0 18px 55px rgba(0,0,0,.22)}}.log-head{{display:flex;align-items:flex-start;justify-content:space-between;gap:14px}}.log-head h2{{margin:0 0 6px;font-size:24px}}.badge{{white-space:nowrap;border-radius:999px;padding:7px 11px;font-weight:900}}.badge.ok{{background:#14532d;color:#bbf7d0;border:1px solid #22c55e}}.badge.warn{{background:#451a03;color:#fed7aa;border:1px solid #f97316}}.meta{{margin-top:10px;color:#cbd5e1}}code{{direction:ltr;display:block;background:#0b1220;border:1px solid #334155;border-radius:12px;padding:10px;color:#fde68a;margin-top:6px;white-space:normal}}.stats{{display:flex;gap:12px;flex-wrap:wrap;color:#cbd5e1;margin:12px 0}}.stats span{{background:#0b1220;border:1px solid #334155;border-radius:999px;padding:8px 10px}}pre{{direction:ltr;text-align:left;white-space:pre-wrap;background:#020617;color:#fde68a;border:1px solid #334155;border-radius:16px;padding:14px;overflow:auto;max-height:420px;font-size:13px;line-height:1.55}}@media(max-width:700px){{body{{padding:14px}}.hero{{padding:18px;border-radius:22px}}h1{{font-size:28px}}.log-head{{display:block}}}}
</style></head><body><div class="wrap"><div class="hero"><a href="/admin">⬅ חזרה לפאנל</a><h1>📜 מרכז לוגים</h1><p>צפייה נוחה בלוגי האדמין, עדכונים ופעולות מערכת בלי לפתוח קבצים ידנית במחשב. המסך לקריאה בלבד ולא מוחק או משנה שום דבר.</p><div class="actions"><a href="/admin/logs">רענן</a><a class="secondary" href="/admin/logs?lines=250">הצג 250 שורות</a><a class="secondary" href="/admin/logs?lines=500">הצג 500 שורות</a><a class="secondary" href="/admin/logs.json" target="_blank">JSON טכני</a><a class="secondary" href="/admin/self-test">בדיקה עצמית</a></div></div>{''.join(cards)}</div></body></html>
'''
def query_watchdog_task():
if os.name != "nt":
return {"supported": False, "installed": False, "ok": False, "details": "Task Scheduler זמין רק על Windows."}
try:
completed = subprocess.run(["schtasks", "/Query", "/TN", WATCHDOG_TASK_NAME, "/FO", "LIST"], capture_output=True, text=True, shell=False, timeout=12)
output = (completed.stdout or completed.stderr or "").strip()
return {"supported": True, "installed": completed.returncode == 0, "ok": completed.returncode == 0, "details": output}
except Exception as exc:
return {"supported": True, "installed": False, "ok": False, "details": repr(exc)}
def watchdog_status(include_log=True, include_task=True):
generated = build_watchdog_code(os.path.abspath(ADMIN_SERVER_PATH), sys.executable or "python", os.path.dirname(os.path.abspath(ADMIN_SERVER_PATH)) or BASE_DIR)
source_ok, source_info = validate_watchdog_source(generated)
file_exists = os.path.exists(WATCHDOG_SCRIPT_PATH)
file_compiles = False
file_info = "לא נכתב עדיין"
if file_exists:
try:
py_compile.compile(WATCHDOG_SCRIPT_PATH, doraise=True)
file_compiles = True
file_info = "OK"
except Exception as exc:
file_info = repr(exc)
task = query_watchdog_task() if include_task else {"supported": None, "installed": None, "ok": None, "details": "not queried"}
return {
"source_ok": source_ok,
"source_status": source_info,
"script_path": WATCHDOG_SCRIPT_PATH,
"script_exists": file_exists,
"script_compil