import os import json from datetime import datetime import random import string import httpx import base64 def create_timestamped_folder(base_dir="reports"): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] rand_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=3)) folder_name = f"{timestamp}_{rand_suffix}" save_dir = os.path.join(base_dir, folder_name) os.makedirs(save_dir, exist_ok=True) return save_dir def save_string_as_md(raw_string, save_dir, filename="report.md"): try: file_path = os.path.join(save_dir, filename) try: formatted_str = raw_string.encode("utf-8").decode("unicode_escape") except Exception: formatted_str = raw_string with open(file_path, "w", encoding="utf-8") as f: f.write(formatted_str) print(f"✅[{filename}] Report saved to {file_path}") except Exception as e: print(f"❌ [{filename}] Failed to save report: {e}") def default_encoder(obj): try: return obj.__dict__ except Exception: return str(obj) def save_json(data, save_dir, filename="output.json"): try: file_path = os.path.join(save_dir, filename) with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False, default=default_encoder,) print(f"✅[{filename}] JSON saved to {file_path}") except Exception as e: print(f"❌ [{filename}]Failed to save json: {e}") async def convert_keyframes_to_base64(ts_result): print("[convert_keyframes_to_base64] start") base64_list = [] async with httpx.AsyncClient() as client: for item in ts_result.get("results", []): url = item.get("keyframe_file") if not url: continue try: resp = await client.get(url, timeout=60) resp.raise_for_status() b64_str = base64.b64encode(resp.content).decode("utf-8") base64_list.append(b64_str) except Exception as e: print(f"⚠️ [convert_keyframes_to_base64]errors {url}: {e}") print("[convert_keyframes_to_base64] done") return base64_list