import os
import sys
import glob
import time
import json
import argparse
import mimetypes
from pathlib import Path
from dotenv import load_dotenv
import google.generativeai as genai
from google.generativeai.types import HarmCategory, HarmBlockThreshold
# Load environment variables
env_path = Path(__file__).parent / "ai_summary.env"
load_dotenv(dotenv_path=env_path)
API_KEY = os.getenv("GEMINI_API_KEY")
if not API_KEY:
print("Error: GEMINI_API_KEY environment variable not found.")
print("Please create a .env file with your API key or set it in your environment.")
sys.exit(1)
genai.configure(api_key=API_KEY)
# Configuration
# Using Gemini 1.5 Flash for speed and cost-efficiency with video
MODEL_NAME = 'gemini-1.5-flash'
VIDEO_EXTENSIONS = {'.mp4', '.mov', '.avi', '.mkv', '.webm', '.flv'}
def setup_model():
# Adjust safety settings to allow processing of the described content
# Note: The API may still block content based on its own internal filters.
safety_settings = {
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
}
generation_config = {
"temperature": 0.4,
"response_mime_type": "application/json",
}
model = genai.GenerativeModel(
model_name=MODEL_NAME,
safety_settings=safety_settings,
generation_config=generation_config
)
return model
def upload_video(video_path):
print(f"Uploading {video_path.name}...")
try:
video_file = genai.upload_file(path=video_path)
print(f"Upload complete: {video_file.name}")
# Wait for processing
while video_file.state.name == "PROCESSING":
print('.', end='', flush=True)
time.sleep(2)
video_file = genai.get_file(video_file.name)
if video_file.state.name == "FAILED":
print(f"\nProcessing failed for {video_path.name}")
return None
print(f"\nVideo is ready.")
return video_file
except Exception as e:
print(f"Error uploading file: {e}")
return None
def analyze_video(model, video_file):
prompt = """
Analyze this video and provide a summary of the actions occurring in it.
Return a JSON object with the following fields:
- \"summary\": A detailed text description of what happens in the video.
- \"contains_only_blowjobs_no_sex\": Boolean (true/false). Set to true ONLY if the video contains oral sex (blowjobs) but DOES NOT contain penetration sex (vaginal or anal).
"""
try:
response = model.generate_content([video_file, prompt])
return json.loads(response.text)
except Exception as e:
print(f"Error generating content: {e}")
return None
def update_html_report(report_path, video_name, summary, flag):
# Simple append logic for HTML
# Check if file exists to write header
file_exists = os.path.exists(report_path)
with open(report_path, 'a', encoding='utf-8') as f:
if not file_exists:
f.write("""
Video Analysis Report
| File Name |
Summary |
Criteria Met (Only BJ, No Sex) |
""")
row_class = ' class="flagged"' if flag else ""
f.write(f" \n | {video_name} | \n {summary} | \n {flag}\n
\n")
def close_html_report(report_path):
if os.path.exists(report_path):
with open(report_path, 'a', encoding='utf-8') as f:
f.write("
\n\n")
def main():
parser = argparse.ArgumentParser(description="Analyze videos for specific content.")
parser.add_argument("directory", help="Target directory containing videos")
args = parser.parse_args()
target_dir = Path(args.directory)
if not target_dir.is_dir():
print(f"Directory not found: {target_dir}")
sys.exit(1)
model = setup_model()
files_to_delete = []
# List all video files
video_files = [
f for f in target_dir.iterdir()
if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS
]
print(f"Found {len(video_files)} videos in {target_dir}")
report_path = target_dir / "analysis_report.html"
# Remove old report if exists to start fresh? Or append?
# Let's start fresh for this run
if report_path.exists():
os.remove(report_path)
for video_path in video_files:
print(f"\nProcessing: {video_path.name}")
# Check if summary already exists to skip?
# Requirement implies we run analysis. Let's assume we run on all.
uploaded_file = upload_video(video_path)
if not uploaded_file:
continue
result = analyze_video(model, uploaded_file)
# Cleanup remote file to save storage/quota (optional but good practice)
try:
genai.delete_file(uploaded_file.name)
except:
pass
if result:
summary = result.get("summary", "No summary provided.")
criteria_met = result.get("contains_only_blowjobs_no_sex", False)
# Save text summary
txt_path = video_path.with_suffix('.txt')
with open(txt_path, 'w', encoding='utf-8') as f:
f.write(summary)
print(f"Saved summary to {txt_path.name}")
# Update HTML
update_html_report(report_path, video_path.name, summary, criteria_met)
if criteria_met:
print("--> MATCHES CRITERIA: Only blowjobs, no sex.")
files_to_delete.append(video_path)
else:
print("--> Does not match deletion criteria.")
else:
print("Failed to analyze video content.")
close_html_report(report_path)
print(f"\nAnalysis complete. Report saved to {report_path}")
if files_to_delete:
print("\n" + "="*40)
print(f"Found {len(files_to_delete)} files matching 'Only Blowjobs, No Sex':")
for f in files_to_delete:
print(f"- {f.name}")
print("="*40)
confirm = input("\nDo you want to DELETE these files? (yes/no): ").lower()
if confirm == 'yes':
for f in files_to_delete:
try:
os.remove(f)
# Also remove the generated text file?
# Usually better to keep the summary or delete it too.
# User asked to delete the files (implying videos).
# I'll delete the video.
print(f"Deleted: {f.name}")
except OSError as e:
print(f"Error deleting {f.name}: {e}")
else:
print("Deletion cancelled.")
else:
print("\nNo files matched the deletion criteria.")
if __name__ == "__main__":
main()