|
#!/usr/bin/env python3 |
|
""" |
|
Find Gemini meeting notes in Google Drive, move them to a specified folder, |
|
and add them as sources to a NotebookLM notebook. |
|
""" |
|
|
|
import argparse |
|
import json |
|
import os |
|
import re |
|
import subprocess |
|
import sys |
|
|
|
|
|
def build_gws_env(profile): |
|
home = os.path.expanduser("~") |
|
env = os.environ.copy() |
|
env["CLOUDSDK_CONFIG"] = os.path.join(home, f".config/gcloud-{profile}") |
|
env["GOOGLE_WORKSPACE_CLI_CONFIG_DIR"] = os.path.join(home, f".config/gws-{profile}") |
|
env["GWS_CONFIG_DIR"] = os.path.join(home, f".config/gws-{profile}") |
|
return env |
|
|
|
|
|
def run_gws(args, profile): |
|
env = build_gws_env(profile) |
|
result = subprocess.run( |
|
["gws"] + args, capture_output=True, text=True, env=env |
|
) |
|
if result.returncode != 0: |
|
print(f"gws error: {result.stderr.strip()}", file=sys.stderr) |
|
sys.exit(1) |
|
try: |
|
return json.loads(result.stdout) |
|
except json.JSONDecodeError: |
|
print(f"Failed to parse gws output: {result.stdout[:300]}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
|
|
def run_nlm(args): |
|
result = subprocess.run( |
|
["nlm"] + args, capture_output=True, text=True |
|
) |
|
if result.returncode != 0: |
|
print(f"nlm error: {result.stderr.strip()}", file=sys.stderr) |
|
print(result.stdout, file=sys.stderr) |
|
sys.exit(1) |
|
return result |
|
|
|
|
|
def resolve_drive_path(path, profile): |
|
"""Walk a slash-separated Drive path from root to get the folder ID.""" |
|
segments = [s for s in path.strip("/").split("/") if s] |
|
if not segments: |
|
print("Error: destination path is empty", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
parent_id = "root" |
|
for segment in segments: |
|
escaped = segment.replace("\\", "\\\\").replace("'", "\\'") |
|
q = ( |
|
f"name = '{escaped}' and " |
|
f"mimeType = 'application/vnd.google-apps.folder' and " |
|
f"'{parent_id}' in parents and " |
|
f"trashed = false" |
|
) |
|
data = run_gws( |
|
["drive", "files", "list", |
|
"--params", json.dumps({"q": q, "fields": "files(id,name)", "pageSize": 1})], |
|
profile, |
|
) |
|
files = data.get("files", []) |
|
if not files: |
|
print(f"Error: folder '{segment}' not found under parent {parent_id}", file=sys.stderr) |
|
sys.exit(1) |
|
parent_id = files[0]["id"] |
|
print(f" Resolved '{segment}' -> {parent_id}") |
|
|
|
return parent_id |
|
|
|
|
|
def find_notes(subject, dest_folder_id, profile): |
|
"""Find Google Docs whose name contains the subject, excluding the destination folder.""" |
|
escaped = subject.replace("\\", "\\\\").replace("'", "\\'") |
|
q = ( |
|
f"name contains '{escaped}' and " |
|
f"mimeType = 'application/vnd.google-apps.document' and " |
|
f"trashed = false" |
|
) |
|
if dest_folder_id: |
|
q = ( |
|
f"name contains '{escaped}' and " |
|
f"mimeType = 'application/vnd.google-apps.document' and " |
|
f"not '{dest_folder_id}' in parents and " |
|
f"trashed = false" |
|
) |
|
data = run_gws( |
|
["drive", "files", "list", |
|
"--params", json.dumps({ |
|
"q": q, |
|
"fields": "files(id,name,parents)", |
|
"pageSize": 100, |
|
})], |
|
profile, |
|
) |
|
files = data.get("files", []) |
|
return [f for f in files if subject.lower() in f.get("name", "").lower()] |
|
|
|
|
|
def move_file(file_id, current_parents, dest_folder_id, profile): |
|
params = { |
|
"fileId": file_id, |
|
"addParents": dest_folder_id, |
|
} |
|
if current_parents: |
|
params["removeParents"] = ",".join(current_parents) |
|
run_gws(["drive", "files", "update", "--params", json.dumps(params)], profile) |
|
|
|
|
|
def create_shortcut(file_id, file_name, dest_folder_id, profile): |
|
body = { |
|
"name": file_name, |
|
"mimeType": "application/vnd.google-apps.shortcut", |
|
"parents": [dest_folder_id], |
|
"shortcutDetails": { |
|
"targetId": file_id, |
|
}, |
|
} |
|
run_gws(["drive", "files", "create", "--json", json.dumps(body)], profile) |
|
|
|
|
|
def find_existing_shortcuts(dest_folder_id, profile): |
|
"""Return the set of target IDs that already have shortcuts in the dest folder.""" |
|
q = ( |
|
f"mimeType = 'application/vnd.google-apps.shortcut' and " |
|
f"'{dest_folder_id}' in parents and " |
|
f"trashed = false" |
|
) |
|
data = run_gws( |
|
["drive", "files", "list", |
|
"--params", json.dumps({ |
|
"q": q, |
|
"fields": "files(shortcutDetails)", |
|
"pageSize": 100, |
|
})], |
|
profile, |
|
) |
|
return { |
|
f["shortcutDetails"]["targetId"] |
|
for f in data.get("files", []) |
|
if f.get("shortcutDetails", {}).get("targetId") |
|
} |
|
|
|
|
|
def create_notebook(title): |
|
result = run_nlm(["notebook", "create", title, "--json"]) |
|
try: |
|
data = json.loads(result.stdout) |
|
nb_id = data.get("id") or data.get("notebook_id") or data.get("project_id") |
|
if not nb_id: |
|
for key, val in data.items(): |
|
if "id" in key.lower() and isinstance(val, str): |
|
nb_id = val |
|
break |
|
if nb_id: |
|
return nb_id |
|
except json.JSONDecodeError: |
|
pass |
|
|
|
match = re.search(r"[a-zA-Z0-9_-]{20,}", result.stdout) |
|
if match: |
|
return match.group(0) |
|
|
|
print(f"Could not extract notebook ID from: {result.stdout[:300]}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
|
|
def count_words_in_doc(doc_id, profile): |
|
"""Fetch a Google Doc via the Docs API and return its word count across all tabs.""" |
|
env = build_gws_env(profile) |
|
result = subprocess.run( |
|
["gws", "docs", "documents", "get", |
|
"--params", json.dumps({ |
|
"documentId": doc_id, |
|
"includeTabsContent": True, |
|
})], |
|
capture_output=True, text=True, env=env, |
|
) |
|
if result.returncode != 0: |
|
print(f"Warning: could not fetch doc {doc_id}: {result.stderr.strip()}", file=sys.stderr) |
|
return None |
|
try: |
|
doc = json.loads(result.stdout) |
|
except json.JSONDecodeError: |
|
print(f"Warning: could not parse doc {doc_id}", file=sys.stderr) |
|
return None |
|
|
|
text_parts = [] |
|
|
|
def extract_text(content_list): |
|
for elem in content_list: |
|
if "paragraph" in elem: |
|
for e in elem["paragraph"].get("elements", []): |
|
text = e.get("textRun", {}).get("content", "") |
|
if text: |
|
text_parts.append(text) |
|
if "table" in elem: |
|
for row in elem["table"].get("tableRows", []): |
|
for cell in row.get("tableCells", []): |
|
extract_text(cell.get("content", [])) |
|
if "tableOfContents" in elem: |
|
extract_text(elem["tableOfContents"].get("content", [])) |
|
|
|
for tab in doc.get("tabs", []): |
|
body = tab.get("documentTab", {}).get("body", {}) |
|
extract_text(body.get("content", [])) |
|
|
|
return len(" ".join(text_parts).split()) |
|
|
|
|
|
def list_sources(notebook_id): |
|
"""Return the list of existing sources in a notebook (oldest first).""" |
|
result = run_nlm(["source", "list", notebook_id, "--json"]) |
|
try: |
|
return json.loads(result.stdout) |
|
except json.JSONDecodeError: |
|
print(f"Failed to parse source list: {result.stdout[:300]}", file=sys.stderr) |
|
sys.exit(1) |
|
|
|
|
|
def delete_sources(source_ids): |
|
"""Delete sources by ID.""" |
|
run_nlm(["source", "delete"] + source_ids + ["--confirm"]) |
|
|
|
|
|
def ensure_source_capacity(notebook_id, needed, max_sources): |
|
"""Check capacity and prompt to remove oldest sources if needed. Returns True if ready.""" |
|
existing = list_sources(notebook_id) |
|
current = len(existing) |
|
available = max_sources - current |
|
|
|
if needed <= available: |
|
return True |
|
|
|
to_remove = needed - available |
|
oldest = existing[:to_remove] |
|
|
|
print(f"\nNotebook has {current}/{max_sources} sources. Adding {needed} would exceed the limit.") |
|
print(f"The {to_remove} oldest source(s) would be removed:") |
|
for src in oldest: |
|
print(f" - {src.get('title', src.get('id', '?'))}") |
|
|
|
answer = input("\nRemove these sources to make room? [y/N] ").strip().lower() |
|
if answer != "y": |
|
print("Aborted.") |
|
sys.exit(0) |
|
|
|
print(f"Removing {to_remove} oldest source(s) ...") |
|
delete_sources([src["id"] for src in oldest]) |
|
return True |
|
|
|
|
|
def add_source(notebook_id, drive_doc_id): |
|
run_nlm(["source", "add", notebook_id, "--drive", drive_doc_id]) |
|
|
|
|
|
def main(): |
|
parser = argparse.ArgumentParser( |
|
description="Find Gemini meeting notes, move to a Drive folder, and add to NotebookLM.", |
|
) |
|
parser.add_argument( |
|
"-s", "--subject", required=True, |
|
help="Meeting subject to match (e.g. 'Team-A Weekly Team Sync')", |
|
) |
|
parser.add_argument( |
|
"-d", "--dest", default=None, |
|
help="Destination path in Google Drive (e.g. 'accounts/TeamA'). If omitted, docs are added to NotebookLM without moving or creating shortcuts.", |
|
) |
|
parser.add_argument( |
|
"-n", "--notebook", default=None, |
|
help="NotebookLM notebook ID. If omitted, a new notebook is created.", |
|
) |
|
parser.add_argument( |
|
"--gws-profile", default="redhat", |
|
help="GWS config profile name (default: redhat)", |
|
) |
|
parser.add_argument( |
|
"--not-owner", action="store_true", |
|
help="Create shortcuts instead of moving files (use when you don't own the docs). Requires --dest.", |
|
) |
|
parser.add_argument( |
|
"--max-sources", type=int, default=300, |
|
help="Maximum number of sources allowed in the notebook (default: 300)", |
|
) |
|
parser.add_argument( |
|
"--max-words", type=int, default=None, |
|
help="Maximum words per source (e.g. 500000). Docs exceeding this are skipped.", |
|
) |
|
parser.add_argument( |
|
"--dry-run", action="store_true", |
|
help="Show what would be done without making changes", |
|
) |
|
args = parser.parse_args() |
|
|
|
if args.not_owner and not args.dest: |
|
parser.error("--not-owner requires --dest") |
|
|
|
dest_folder_id = None |
|
if args.dest: |
|
print(f"Resolving Drive path: {args.dest}") |
|
dest_folder_id = resolve_drive_path(args.dest, args.gws_profile) |
|
print(f"Destination folder ID: {dest_folder_id}\n") |
|
|
|
print(f"Searching for docs matching: '{args.subject}'") |
|
if args.not_owner: |
|
docs = find_notes(args.subject, None, args.gws_profile) |
|
existing = find_existing_shortcuts(dest_folder_id, args.gws_profile) |
|
docs = [d for d in docs if d["id"] not in existing] |
|
else: |
|
docs = find_notes(args.subject, dest_folder_id, args.gws_profile) |
|
|
|
if not docs: |
|
print("No matching documents found.") |
|
return |
|
|
|
if args.not_owner: |
|
action = "shortcut" |
|
elif args.dest: |
|
action = "move" |
|
else: |
|
action = "add" |
|
print(f"Found {len(docs)} document(s) to {action}:") |
|
for doc in docs: |
|
print(f" - {doc['name']} ({doc['id']})") |
|
print() |
|
|
|
if args.max_words: |
|
print(f"Checking word counts (limit: {args.max_words:,} words) ...") |
|
eligible = [] |
|
for doc in docs: |
|
wc = count_words_in_doc(doc["id"], args.gws_profile) |
|
if wc is None: |
|
print(f" Warning: could not check {doc['name']}, including anyway") |
|
eligible.append(doc) |
|
elif wc > args.max_words: |
|
print(f" SKIPPED: {doc['name']} ({wc:,} words exceeds {args.max_words:,} limit)") |
|
else: |
|
print(f" OK: {doc['name']} ({wc:,} words)") |
|
eligible.append(doc) |
|
if len(eligible) < len(docs): |
|
print(f"\n{len(docs) - len(eligible)} doc(s) skipped for exceeding word limit.") |
|
docs = eligible |
|
if not docs: |
|
print("No eligible documents remaining.") |
|
return |
|
print() |
|
|
|
if args.dry_run: |
|
print(f"[DRY RUN] Would {action} the above documents and add them to NotebookLM.") |
|
return |
|
|
|
notebook_id = args.notebook |
|
if not notebook_id: |
|
print(f"Creating new NotebookLM notebook: '{args.subject}'") |
|
notebook_id = create_notebook(args.subject) |
|
print(f"Created notebook: {notebook_id}\n") |
|
else: |
|
print(f"Using notebook: {notebook_id}\n") |
|
|
|
ensure_source_capacity(notebook_id, len(docs), args.max_sources) |
|
|
|
for doc in docs: |
|
doc_id = doc["id"] |
|
doc_name = doc["name"] |
|
parents = doc.get("parents", []) |
|
|
|
print(f"Processing: {doc_name}") |
|
if args.not_owner: |
|
print(f" Creating shortcut in {args.dest} ...") |
|
create_shortcut(doc_id, doc_name, dest_folder_id, args.gws_profile) |
|
elif args.dest: |
|
print(f" Moving to {args.dest} ...") |
|
move_file(doc_id, parents, dest_folder_id, args.gws_profile) |
|
print(f" Adding to NotebookLM ...") |
|
add_source(notebook_id, doc_id) |
|
print(f" Done.") |
|
|
|
print(f"\nComplete. Processed {len(docs)} document(s).") |
|
print(f"Notebook ID: {notebook_id}") |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |