|
1
|
"""Google Meet recording source using the gws CLI (googleworkspace/cli). |
|
2
|
|
|
3
|
Fetches Meet recordings and companion transcripts from Google Drive |
|
4
|
via the `gws` CLI tool. |
|
5
|
|
|
6
|
Requires: npm install -g @googleworkspace/cli |
|
7
|
Auth: gws auth login (interactive) or GOOGLE_WORKSPACE_CLI_CREDENTIALS_FILE (headless) |
|
8
|
""" |
|
9
|
|
|
10
|
import json |
|
11
|
import logging |
|
12
|
import re |
|
13
|
import shutil |
|
14
|
import subprocess |
|
15
|
from pathlib import Path |
|
16
|
from typing import List, Optional |
|
17
|
|
|
18
|
from video_processor.sources.base import BaseSource, SourceFile |
|
19
|
from video_processor.sources.gws_source import _run_gws |
|
20
|
|
|
21
|
logger = logging.getLogger(__name__) |
|
22
|
|
|
23
|
|
|
24
|
class MeetRecordingSource(BaseSource): |
|
25
|
""" |
|
26
|
Fetch Google Meet recordings and transcripts from Google Drive via the gws CLI. |
|
27
|
|
|
28
|
Meet stores recordings as MP4 files in Drive (typically in a "Meet Recordings" |
|
29
|
folder) and auto-generated transcripts as Google Docs. |
|
30
|
|
|
31
|
Usage: |
|
32
|
source = MeetRecordingSource() |
|
33
|
source.authenticate() |
|
34
|
recordings = source.list_videos() |
|
35
|
source.download_all(recordings, Path("./recordings")) |
|
36
|
|
|
37
|
# Fetch transcript for a specific recording |
|
38
|
transcript = source.fetch_transcript("Meet Recording 2026-03-07") |
|
39
|
""" |
|
40
|
|
|
41
|
def __init__(self, drive_folder_id: Optional[str] = None): |
|
42
|
self.drive_folder_id = drive_folder_id |
|
43
|
|
|
44
|
def authenticate(self) -> bool: |
|
45
|
"""Check if gws CLI is installed and authenticated.""" |
|
46
|
if not shutil.which("gws"): |
|
47
|
logger.error("gws CLI not found. Install with: npm install -g @googleworkspace/cli") |
|
48
|
return False |
|
49
|
try: |
|
50
|
_run_gws(["auth", "status"], timeout=10) |
|
51
|
return True |
|
52
|
except (RuntimeError, subprocess.TimeoutExpired): |
|
53
|
logger.error("gws not authenticated. Run: gws auth login") |
|
54
|
return False |
|
55
|
|
|
56
|
def list_videos( |
|
57
|
self, |
|
58
|
folder_id: Optional[str] = None, |
|
59
|
folder_path: Optional[str] = None, |
|
60
|
patterns: Optional[List[str]] = None, |
|
61
|
) -> List[SourceFile]: |
|
62
|
"""List Google Meet recordings in Drive. |
|
63
|
|
|
64
|
Searches for MP4 files with 'Meet Recording' in the name. If a |
|
65
|
drive_folder_id is set, restricts search to that folder. |
|
66
|
Also discovers companion transcript docs for each recording. |
|
67
|
""" |
|
68
|
target_folder = folder_id or self.drive_folder_id |
|
69
|
files: List[SourceFile] = [] |
|
70
|
|
|
71
|
# Build the Drive search query for Meet recordings |
|
72
|
q_parts = [ |
|
73
|
"mimeType='video/mp4'", |
|
74
|
"name contains 'Meet Recording'", |
|
75
|
"trashed=false", |
|
76
|
] |
|
77
|
if target_folder: |
|
78
|
q_parts.append(f"'{target_folder}' in parents") |
|
79
|
|
|
80
|
params = { |
|
81
|
"q": " and ".join(q_parts), |
|
82
|
"fields": "files(id,name,mimeType,size,modifiedTime)", |
|
83
|
"pageSize": 50, |
|
84
|
"orderBy": "modifiedTime desc", |
|
85
|
} |
|
86
|
|
|
87
|
try: |
|
88
|
result = _run_gws( |
|
89
|
[ |
|
90
|
"drive", |
|
91
|
"files", |
|
92
|
"list", |
|
93
|
"--params", |
|
94
|
json.dumps(params), |
|
95
|
], |
|
96
|
timeout=60, |
|
97
|
) |
|
98
|
except RuntimeError as e: |
|
99
|
logger.error(f"Failed to list Meet recordings: {e}") |
|
100
|
return [] |
|
101
|
|
|
102
|
recordings = result.get("files", []) |
|
103
|
for item in recordings: |
|
104
|
size = item.get("size") |
|
105
|
files.append( |
|
106
|
SourceFile( |
|
107
|
name=item.get("name", "Meet Recording"), |
|
108
|
id=item.get("id", ""), |
|
109
|
size_bytes=int(size) if size else None, |
|
110
|
mime_type=item.get("mimeType", "video/mp4"), |
|
111
|
modified_at=item.get("modifiedTime"), |
|
112
|
) |
|
113
|
) |
|
114
|
|
|
115
|
# Also search for auto-generated transcript docs |
|
116
|
transcript_params = { |
|
117
|
"q": " and ".join( |
|
118
|
[ |
|
119
|
"mimeType='application/vnd.google-apps.document'", |
|
120
|
"(name contains 'Transcript' or name contains 'Meeting notes')", |
|
121
|
"trashed=false", |
|
122
|
] |
|
123
|
+ ([f"'{target_folder}' in parents"] if target_folder else []) |
|
124
|
), |
|
125
|
"fields": "files(id,name,mimeType,modifiedTime)", |
|
126
|
"pageSize": 50, |
|
127
|
"orderBy": "modifiedTime desc", |
|
128
|
} |
|
129
|
|
|
130
|
try: |
|
131
|
transcript_result = _run_gws( |
|
132
|
[ |
|
133
|
"drive", |
|
134
|
"files", |
|
135
|
"list", |
|
136
|
"--params", |
|
137
|
json.dumps(transcript_params), |
|
138
|
], |
|
139
|
timeout=60, |
|
140
|
) |
|
141
|
transcript_files = transcript_result.get("files", []) |
|
142
|
logger.info( |
|
143
|
f"Found {len(recordings)} recording(s) and " |
|
144
|
f"{len(transcript_files)} transcript doc(s) in Drive" |
|
145
|
) |
|
146
|
except RuntimeError as e: |
|
147
|
logger.debug(f"Transcript search failed: {e}") |
|
148
|
|
|
149
|
if not files: |
|
150
|
logger.warning("No Google Meet recordings found in Drive") |
|
151
|
|
|
152
|
logger.info(f"Found {len(files)} Meet recording(s)") |
|
153
|
return files |
|
154
|
|
|
155
|
def download(self, file: SourceFile, destination: Path) -> Path: |
|
156
|
"""Download a Meet recording from Drive.""" |
|
157
|
destination = Path(destination) |
|
158
|
destination.parent.mkdir(parents=True, exist_ok=True) |
|
159
|
|
|
160
|
# For video files, download binary content via alt=media |
|
161
|
result = _run_gws( |
|
162
|
[ |
|
163
|
"drive", |
|
164
|
"files", |
|
165
|
"get", |
|
166
|
"--params", |
|
167
|
json.dumps({"fileId": file.id, "alt": "media"}), |
|
168
|
], |
|
169
|
timeout=300, |
|
170
|
) |
|
171
|
|
|
172
|
# Write the content — result may be raw binary or a dict wrapper |
|
173
|
raw = result.get("raw", "") if isinstance(result, dict) else str(result) |
|
174
|
destination.write_text(raw, encoding="utf-8") |
|
175
|
logger.info(f"Downloaded {file.name} to {destination}") |
|
176
|
return destination |
|
177
|
|
|
178
|
def fetch_transcript(self, recording_name: str) -> Optional[str]: |
|
179
|
"""Fetch the companion transcript for a Meet recording. |
|
180
|
|
|
181
|
Google Meet creates transcript docs with names that typically match |
|
182
|
the recording date/time. This method searches for the matching |
|
183
|
Google Doc and extracts its text content. |
|
184
|
""" |
|
185
|
transcript_id = self._find_matching_transcript(recording_name) |
|
186
|
if not transcript_id: |
|
187
|
logger.info(f"No matching transcript found for: {recording_name}") |
|
188
|
return None |
|
189
|
|
|
190
|
# Fetch the Google Doc content via the Docs API |
|
191
|
try: |
|
192
|
result = _run_gws( |
|
193
|
[ |
|
194
|
"docs", |
|
195
|
"documents", |
|
196
|
"get", |
|
197
|
"--params", |
|
198
|
json.dumps({"documentId": transcript_id}), |
|
199
|
], |
|
200
|
timeout=60, |
|
201
|
) |
|
202
|
except RuntimeError as e: |
|
203
|
logger.warning(f"Failed to fetch transcript doc {transcript_id}: {e}") |
|
204
|
return None |
|
205
|
|
|
206
|
# Extract text from the Docs API structural response |
|
207
|
body = result.get("body", {}) |
|
208
|
text_parts: list[str] = [] |
|
209
|
for element in body.get("content", []): |
|
210
|
paragraph = element.get("paragraph", {}) |
|
211
|
for pe in paragraph.get("elements", []): |
|
212
|
text_run = pe.get("textRun", {}) |
|
213
|
text = text_run.get("content", "") |
|
214
|
if text.strip(): |
|
215
|
text_parts.append(text) |
|
216
|
|
|
217
|
if not text_parts: |
|
218
|
logger.warning(f"Transcript doc {transcript_id} had no extractable text") |
|
219
|
return None |
|
220
|
|
|
221
|
return "".join(text_parts) |
|
222
|
|
|
223
|
def _find_matching_transcript(self, recording_name: str) -> Optional[str]: |
|
224
|
"""Search Drive for a transcript doc that matches a recording name. |
|
225
|
|
|
226
|
Meet recordings are typically named like: |
|
227
|
"Meet Recording 2026-03-07T14:30:00" |
|
228
|
And transcripts are named like: |
|
229
|
"Meeting Transcript 2026-03-07" or "2026-03-07 - Transcript" |
|
230
|
|
|
231
|
This extracts the date portion and searches for matching transcript docs. |
|
232
|
""" |
|
233
|
# Extract a date string from the recording name (YYYY-MM-DD pattern) |
|
234
|
date_match = re.search(r"\d{4}-\d{2}-\d{2}", recording_name) |
|
235
|
date_str = date_match.group(0) if date_match else recording_name |
|
236
|
|
|
237
|
# Search for transcript docs matching the date |
|
238
|
search_query = " and ".join( |
|
239
|
[ |
|
240
|
"mimeType='application/vnd.google-apps.document'", |
|
241
|
f"name contains '{date_str}'", |
|
242
|
"(name contains 'Transcript' or name contains 'transcript' " |
|
243
|
"or name contains 'Meeting notes')", |
|
244
|
"trashed=false", |
|
245
|
] |
|
246
|
) |
|
247
|
if self.drive_folder_id: |
|
248
|
search_query += f" and '{self.drive_folder_id}' in parents" |
|
249
|
|
|
250
|
try: |
|
251
|
result = _run_gws( |
|
252
|
[ |
|
253
|
"drive", |
|
254
|
"files", |
|
255
|
"list", |
|
256
|
"--params", |
|
257
|
json.dumps( |
|
258
|
{ |
|
259
|
"q": search_query, |
|
260
|
"fields": "files(id,name,modifiedTime)", |
|
261
|
"pageSize": 5, |
|
262
|
"orderBy": "modifiedTime desc", |
|
263
|
} |
|
264
|
), |
|
265
|
], |
|
266
|
timeout=60, |
|
267
|
) |
|
268
|
except RuntimeError as e: |
|
269
|
logger.debug(f"Transcript search failed for '{date_str}': {e}") |
|
270
|
return None |
|
271
|
|
|
272
|
files = result.get("files", []) |
|
273
|
if not files: |
|
274
|
logger.debug(f"No transcript docs found matching '{date_str}'") |
|
275
|
return None |
|
276
|
|
|
277
|
# Return the most recently modified match |
|
278
|
best = files[0] |
|
279
|
logger.info(f"Matched transcript: {best.get('name')} for recording: {recording_name}") |
|
280
|
return best.get("id") |
|
281
|
|