#!/usr/bin/env python3 from __future__ import annotations import json import logging import re import sys from pathlib import Path from _gemini_runner import GeminiRunner, load_repo_module BASE = load_repo_module("qwen_multi_event_base", "qwen/generate_multi_event_annotations.py") def strip_detailed_caption_from_global_prompt(prompt: str) -> str: cleaned = prompt cleaned = cleaned.replace( '- The detailed caption should be about 500-800 words.\n', "", ) cleaned = cleaned.replace( '- The detailed caption may synthesize and reuse the structured fields plus stable recurring evidence from the event summaries, but it must not turn into a segment-by-segment narration.\n', "", ) cleaned = re.sub( r',\n(\s*)"detailed_caption": "About 500-800 words\."\n', "\n", cleaned, ) return cleaned _ORIGINAL_BUILD_GLOBAL_CONTEXT_PROMPT = BASE.build_global_context_prompt _ORIGINAL_BUILD_GLOBAL_REFINE_PROMPT = BASE.build_global_refine_prompt _ORIGINAL_COERCE_GLOBAL_PAYLOAD = BASE.coerce_global_payload def _gemini_build_global_context_prompt(*args, **kwargs): prompt = _ORIGINAL_BUILD_GLOBAL_CONTEXT_PROMPT(*args, **kwargs) return ( f"{prompt}\n\n" "IMPORTANT FOR GEMINI:\n" "- Return one single JSON object for the entire stitched video.\n" "- Never return a JSON array.\n" "- Never return one object per clip, per segment, or per event.\n" "- The stitched video may contain multiple chronological event clips, but this context pass must summarize the whole stitched video into one object only.\n" "- Do not return object-detection output, tracking output, grounding output, OCR output, or visual tool output.\n" "- Never return keys such as box_2d, bbox, label, labels, mask, polygon, points, score, or tracking_id unless the requested schema explicitly asks for them.\n" "- Never return timestamped clip-local records such as event_clip_id, start_time_seconds, or end_time_seconds in this context pass.\n" "- If uncertain, still fill the requested schema directly with high-confidence whole-video annotations rather than switching to detection-style results.\n" ) def _gemini_build_global_refine_prompt(*args, **kwargs): return strip_detailed_caption_from_global_prompt( _ORIGINAL_BUILD_GLOBAL_REFINE_PROMPT(*args, **kwargs) ) def _gemini_coerce_global_payload(*args, **kwargs): payload = _ORIGINAL_COERCE_GLOBAL_PAYLOAD(*args, **kwargs) payload.pop("detailed_caption", None) return payload BASE.build_global_context_prompt = _gemini_build_global_context_prompt BASE.build_global_refine_prompt = _gemini_build_global_refine_prompt BASE.coerce_global_payload = _gemini_coerce_global_payload def main() -> int: args = BASE.parse_args() BASE.setup_logging(args.log_level) input_path = Path(args.input).expanduser().resolve() output_path = Path(args.output).expanduser().resolve() camera_motion_guide_path = Path(args.camera_motion_guide).expanduser().resolve() suffixes = BASE.parse_video_suffixes(args.video_suffixes) allow_uuids = BASE.load_uuid_allow_list(args.uuids, args.uuids_file) camera_motion_guide = BASE.load_camera_motion_guide(camera_motion_guide_path) groups = BASE.collect_segment_groups( input_path=input_path, recursive=args.recursive, suffixes=suffixes, allow_uuids=allow_uuids, ) if not groups: raise RuntimeError("No valid segmented videos were found.") uuids = list(groups) if args.limit is not None: uuids = uuids[: args.limit] pending_for_print = uuids if args.resume and output_path.exists(): existing_videos_for_print = BASE.load_existing_output( output_path=output_path, overwrite=False, resume=True, ) pending_for_print = [uuid for uuid in uuids if uuid not in existing_videos_for_print] if args.print_uuids_json: print(json.dumps(pending_for_print, ensure_ascii=False)) return 0 uuids = BASE.apply_uuid_shard( uuids=pending_for_print if args.resume else uuids, num_shards=args.num_shards, shard_index=args.shard_index, ) logging.info("Discovered %d UUID groups.", len(uuids)) for uuid in uuids: logging.info("UUID %s has %d segments.", uuid, len(groups[uuid])) if args.dry_run: logging.info("Dry run complete. Gemini model was not called.") return 0 existing_videos = BASE.load_existing_output( output_path=output_path, overwrite=args.overwrite, resume=args.resume, ) pending_uuids = [uuid for uuid in uuids if uuid not in existing_videos] if not pending_uuids: logging.info("Nothing to do. All requested UUIDs already exist in %s.", output_path) return 0 runner = GeminiRunner( model_path=Path(args.model_path).expanduser().resolve(), device_map=args.device_map, dtype=args.dtype, attn_implementation=args.attn_implementation, trust_remote_code=args.trust_remote_code, fps=args.fps, max_pixels=args.max_pixels, temperature=args.temperature, top_p=args.top_p, top_k=args.top_k, repetition_penalty=args.repetition_penalty, greedy=args.greedy, ) try: failures: list[str] = [] for index, uuid in enumerate(pending_uuids, start=1): logging.info("Processing UUID %s (%d/%d)", uuid, index, len(pending_uuids)) try: video_payload = BASE.annotate_uuid( uuid=uuid, segments=groups[uuid], runner=runner, camera_motion_guide=camera_motion_guide, args=args, ) existing_videos[uuid] = video_payload BASE.write_output(output_path, existing_videos) logging.info("Wrote %s", output_path) except Exception as exc: if not args.continue_on_error: raise logging.exception("Failed to process %s: %s", uuid, exc) failures.append(uuid) if failures: logging.error("Completed with failures: %s", ", ".join(failures)) return 1 logging.info("Finished %d UUIDs.", len(pending_uuids)) return 0 finally: runner.close() if __name__ == "__main__": sys.exit(main())