139 lines
5.1 KiB
Python
Executable File
139 lines
5.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
import csv
|
||
from pathlib import Path
|
||
import sys
|
||
import json
|
||
import shlex
|
||
import subprocess
|
||
|
||
def get_chapters(json_path):
|
||
"""Loads the JSON file and returns all chapters."""
|
||
with open(json_path, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
# The chapters are under video -> chapters
|
||
video = data.get('video', {})
|
||
chapters = video.get('chapters', [])
|
||
|
||
if not chapters:
|
||
print("No chapters found.")
|
||
return []
|
||
|
||
print(f"Found {len(chapters)} chapter(s) in video '{video.get('title')}':\n")
|
||
|
||
for i, chapter in enumerate(chapters, 1):
|
||
start_sec = chapter.get('startMilliseconds', 0) / 1000
|
||
length_sec = chapter.get('lengthMilliseconds', 0) / 1000
|
||
|
||
#print(f"{i}. {chapter.get('description', 'No Title')}")
|
||
#print(f" - Type: {chapter.get('type')}")
|
||
#print(f" - Start: {start_sec}s ({chapter.get('startMilliseconds')}ms)")
|
||
#print(f" - Length: {length_sec}s ({chapter.get('lengthMilliseconds')}ms)")
|
||
#print(f" - Game: {chapter.get('gameDisplayName')} (ID: {chapter.get('gameId')})")
|
||
#print()
|
||
|
||
return chapters
|
||
|
||
def get_video_id_from_chat(json_path: str | Path) -> str | None:
|
||
"""Gets the id inside video:{ } - from the chat json you download"""
|
||
try:
|
||
with open(json_path, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return data.get("video", {}).get("id")
|
||
except Exception as e:
|
||
print(f"Could not read video.id from {json_path}: {e}")
|
||
return None
|
||
|
||
def write_csv(data: list, file_name: str):
|
||
"""Writes a list of dictionaries or objects to a CSV file."""
|
||
|
||
if not data:
|
||
print(f"⚠️ Warning: No data provided to write to {file_name}.")
|
||
return
|
||
|
||
# Determine if the data contains objects that need serialization
|
||
first_item = data[0]
|
||
|
||
if hasattr(first_item, '__dict__') and not isinstance(first_item, (str, int, float, bool)):
|
||
print("ℹ️ Detected object types. Converting objects to dictionary rows for CSV writing.")
|
||
|
||
# Extract attributes/properties into a list of dictionaries
|
||
data_to_write = []
|
||
# We assume all objects in the list have the same structure for simplicity
|
||
# A more robust solution would require knowing the schema.
|
||
for item in data:
|
||
row_dict = {}
|
||
# Export all instance attributes to the dictionary
|
||
for key, value in item.__dict__.items():
|
||
row_dict[key] = value
|
||
data_to_write.append(row_dict)
|
||
|
||
# Since csv.writer doesn't handle dicts directly, we list the keys from the first dict
|
||
fieldnames = list(data_to_write[0].keys())
|
||
|
||
with open(file_name, "w", newline="", encoding="utf-8") as file:
|
||
writer = csv.DictWriter(file, fieldnames=fieldnames)
|
||
|
||
writer.writeheader()
|
||
writer.writerows(data_to_write)
|
||
|
||
else:
|
||
# Assume data is already a list of lists/tuples
|
||
with open(file_name, "w", newline="", encoding="utf-8") as file:
|
||
writer = csv.writer(file)
|
||
|
||
# Write all rows at once
|
||
writer.writerows(data)
|
||
|
||
def run_command(cmd_str: str, progress_prefix: str = "Progress", look_for: list = ["frame=", "time=", "fps=", "Rendering frame"]) -> bool:
|
||
"""Runs a system command, streams its output live, and reports errors on failure."""
|
||
args = shlex.split(cmd_str)
|
||
|
||
# Redirect stderr to stdout to catch all logging/progress in one stream
|
||
process = subprocess.Popen(
|
||
args,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
bufsize=1
|
||
)
|
||
print(f"Executing: {cmd_str[:90]}...")
|
||
|
||
# Maintain a small buffer history to display context if a crash occurs
|
||
output_history = []
|
||
|
||
# Stream the output live to the terminal
|
||
while True:
|
||
line = process.stdout.readline()
|
||
if not line and process.poll() is not None:
|
||
break
|
||
if line:
|
||
clean_line = line.strip()
|
||
output_history.append(clean_line) # Keep history for error reporting
|
||
|
||
# Keep history slim by only keeping the last 20 lines
|
||
if len(output_history) > 20:
|
||
output_history.pop(0)
|
||
|
||
# Only print updates that show progress metrics to keep terminal clean
|
||
if any(metric in clean_line for metric in look_for):
|
||
sys.stdout.write(f"\r[{progress_prefix}] {clean_line}")
|
||
sys.stdout.flush()
|
||
elif "Error" in clean_line or "failed" in clean_line:
|
||
print(f"\n[Alert] {clean_line}")
|
||
|
||
print("\n") # New line after process finishes
|
||
|
||
# Evaluate success status
|
||
success = (process.returncode == 0)
|
||
|
||
if not success:
|
||
print(f"❌ Command failed with exit code: {process.returncode}")
|
||
print("--- Technical Error Details (Last 5 lines of output) ---")
|
||
# Print the last 5 captured lines to show the exact point of failure
|
||
for error_line in output_history[-5:]:
|
||
print(f" > {error_line}")
|
||
print("---------------------------------------------------------")
|
||
|
||
return success
|