cwadayi commited on
Commit
2e07db0
·
verified ·
1 Parent(s): 1240a6f

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +106 -124
app.py CHANGED
@@ -13,14 +13,38 @@ from googleapiclient.http import MediaIoBaseUpload
13
  from googleapiclient.errors import HttpError
14
  from google.oauth2.service_account import Credentials
15
 
16
- # ---------------- Config ----------------
17
  DEFAULT_EXTENT = (119.0, 123.5, 21.5, 25.5) # Taiwan region
18
- SCOPES = ["https://www.googleapis.com/auth/drive.file"] # manage files created by the app
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
19
 
 
 
20
 
21
- # --------------- Helpers ----------------
22
- def _parse_points(csv_text: str) -> List[Tuple[float, float]]:
23
- """Parse lines like '121.5,25.0' (lon,lat). Ignore blank/comment/invalid."""
 
 
24
  pts: List[Tuple[float, float]] = []
25
  if not csv_text:
26
  return pts
@@ -39,10 +63,9 @@ def _parse_points(csv_text: str) -> List[Tuple[float, float]]:
39
  return pts
40
 
41
 
42
- def _make_figure(points: List[Tuple[float, float]], title: str,
43
- extent: Tuple[float, float, float, float] = DEFAULT_EXTENT,
44
- dpi: int = 150) -> str:
45
- """Draw scatter in extent and save to a temp PNG; return file path."""
46
  min_lon, max_lon, min_lat, max_lat = extent
47
  fig, ax = plt.subplots(figsize=(6.5, 6.5), dpi=dpi)
48
  ax.set_title(title or "Map Plot", pad=10)
@@ -62,11 +85,10 @@ def _make_figure(points: List[Tuple[float, float]], title: str,
62
  return out_path
63
 
64
 
65
- def _drive_service_from_env():
66
- """Build Drive service from SERVICE_ACCOUNT_JSON. Returns (service, client_email, err)."""
67
  raw = os.environ.get("SERVICE_ACCOUNT_JSON", "").strip()
68
  if not raw:
69
- return None, None, "SERVICE_ACCOUNT_JSON not found (no secret injected)."
70
  try:
71
  info = json.loads(raw)
72
  except Exception as e:
@@ -80,44 +102,21 @@ def _drive_service_from_env():
80
  return None, None, f"Failed to build Drive service: {e}"
81
 
82
 
83
- def _normalize_folder_id(raw_id: str) -> Optional[str]:
84
- """Accept a bare ID or a full Drive URL; return clean folder ID."""
85
- if not raw_id:
86
- return None
87
- raw = raw_id.strip()
88
- # 從完整 URL 擷取 /folders/<ID>
89
- m = re.search(r"/folders/([A-Za-z0-9_-]+)", raw)
90
- if m:
91
- raw = m.group(1)
92
- # 去掉 query string
93
- raw = raw.split("?")[0]
94
- # 去掉常見尾端符號(半形/全形),避免 '...dRo.' / '...dRo。'
95
- raw = raw.rstrip(" .。;;,,!!??、/\\』」)》)】]}")
96
- # 驗證格式
97
- return raw if re.fullmatch(r"[A-Za-z0-9_-]{20,}", raw) else None
98
-
99
-
100
- def _upload_to_drive(local_path: str, service, folder_id: Optional[str] = None,
101
- make_public_link: bool = False) -> Tuple[str, str]:
102
- """Upload file; return (file_id, webViewLink). Supports Shared Drives."""
103
- fname = os.path.basename(local_path)
104
- meta = {"name": fname}
105
  if folder_id:
106
  meta["parents"] = [folder_id]
107
-
108
- print(f"[Drive] creating file: name={fname}, folder_id={folder_id}") # DEBUG
109
-
110
- with open(local_path, "rb") as f:
111
  media = MediaIoBaseUpload(f, mimetype="image/png", resumable=False)
112
  created = (service.files()
113
  .create(body=meta, media_body=media,
114
  fields="id, webViewLink, webContentLink",
115
  supportsAllDrives=True)
116
  .execute())
117
- file_id = created.get("id", ""); view_link = created.get("webViewLink", "")
118
- print(f"[Drive] created file_id={file_id}, view_link={view_link}") # DEBUG
119
 
120
- if make_public_link and file_id:
121
  try:
122
  service.permissions().create(
123
  fileId=file_id,
@@ -125,120 +124,108 @@ def _upload_to_drive(local_path: str, service, folder_id: Optional[str] = None,
125
  fields="id",
126
  supportsAllDrives=True,
127
  ).execute()
128
- print(f"[Drive] set anyone-with-link permission on {file_id}") # DEBUG
129
  except HttpError as e:
130
- print(f"[Drive] set permission failed: {e}") # DEBUG
131
- return file_id, view_link
132
 
133
-
134
- # --------------- Gradio handlers ---------------
135
- def generate_and_upload(points_text: str, title: str,
136
- min_lon: float, max_lon: float, min_lat: float, max_lat: float,
137
- drive_folder_id_input: str, make_public: bool):
138
- # 1) render figure
139
  try:
140
- points = _parse_points(points_text)
141
- extent = (min_lon, max_lon, min_lat, max_lat)
142
- local_png = _make_figure(points, title, extent=extent)
143
  except Exception as e:
144
  return None, f"❌ Failed to render figure: {e}"
145
 
146
- # 2) drive service
147
- service, client_email, err = _drive_service_from_env()
148
  if service is None:
149
- msg = ("⚠️ Google Drive upload is not configured.\n"
150
- "請在 Settings → Secrets 新增 `SERVICE_ACCOUNT_JSON`(貼完整 JSON)。\n"
151
- "(可選)新增 `DRIVE_FOLDER_ID` 指定資料夾。"
152
- f"\nDebug: {err}")
153
- return local_png, msg
154
 
155
- # 3) folder id(UI 覆蓋 Secret,且會自動清理/解析 URL)
156
- folder_id_env = os.environ.get("DRIVE_FOLDER_ID", "").strip()
157
- raw_folder = (drive_folder_id_input or folder_id_env or "").strip()
158
- folder_id = _normalize_folder_id(raw_folder)
159
 
160
- # DEBUG
161
- print(f"[Drive] client_email={client_email}, using folder_id={folder_id}") # DEBUG
 
162
 
163
- # 4) upload
164
  try:
165
- file_id, view_link = _upload_to_drive(local_png, service, folder_id, make_public)
166
- details = [
167
  "✅ Uploaded to Google Drive.",
 
 
168
  f"• File ID: `{file_id}`",
169
- f"• Folder ID used: `{folder_id or '(none)'}`",
170
  ]
171
  if view_link:
172
- details.append(f"• Open: {view_link}")
173
  if not make_public:
174
- details.append("• Note: 連結權限沿用資料夾/硬碟設定(未自動公開)。")
175
- return local_png, "\n".join(details)
176
  except HttpError as e:
177
- return local_png, f"❌ Upload failed (HttpError): {e}"
178
  except Exception as e:
179
- return local_png, f"❌ Upload failed: {e}"
180
 
181
 
182
  def check_config():
183
- """Quick diagnostics for SERVICE_ACCOUNT_JSON and DRIVE_FOLDER_ID."""
184
  raw = os.environ.get("SERVICE_ACCOUNT_JSON", None)
185
- folder_id_env = os.environ.get("DRIVE_FOLDER_ID", "").strip()
186
- folder_id = _normalize_folder_id(folder_id_env) if folder_id_env else None
 
187
 
188
  if raw is None:
189
- return "❌ 環境變數 `SERVICE_ACCOUNT_JSON` 不存在(Secret 未注入或名稱拼錯)。"
190
- if len(raw.strip()) == 0:
191
- return "❌ `SERVICE_ACCOUNT_JSON` 是空字串。請重新貼完整 JSON。"
192
  try:
193
  info = json.loads(raw)
 
194
  except Exception as e:
195
  return f"❌ `SERVICE_ACCOUNT_JSON` 不是有效 JSON:{e}"
196
- client_email = info.get("client_email")
197
 
198
- # 嘗試讀取資料夾名稱以確認 ID 與讀取權限
199
- folder_line = f"• Folder ID (from Secret): `{folder_id or '(none)'}`"
200
  try:
201
- service, _, err = _drive_service_from_env()
202
  if service and folder_id:
203
  meta = service.files().get(
204
  fileId=folder_id, fields="id, name", supportsAllDrives=True
205
  ).execute()
206
- folder_line += f"(name: {meta.get('name')}"
207
  elif err:
208
- folder_line += f"service build error: {err}"
209
  except HttpError as e:
210
- folder_line += f"(get folder meta failed: {e}"
211
-
212
- return f"✅ 已讀到 SERVICE_ACCOUNT_JSON,client_email = {client_email}\n{folder_line}"
213
 
214
-
215
- def list_folder_contents(drive_folder_id_input: str):
216
- """列出指定 Folder 的前 50 個檔案;若輸入框為空則用 Secret 的 DRIVE_FOLDER_ID。支援貼整個 URL。"""
217
- service, client_email, err = _drive_service_from_env()
218
  if service is None:
219
- return f"❌ 讀取 SERVICE_ACCOUNT_JSON 失敗:{err}"
220
 
221
- folder_id_env = os.environ.get("DRIVE_FOLDER_ID", "").strip()
222
- raw_folder = (drive_folder_id_input or folder_id_env or "").strip()
223
- folder_id = _normalize_folder_id(raw_folder)
224
  if not folder_id:
225
- return "❌ 沒有提供有效的 Folder ID(輸入框與 Secret 都是空的或格式不對)。"
226
 
227
  try:
228
- folder_meta = service.files().get(
229
  fileId=folder_id, fields="id, name, mimeType",
230
  supportsAllDrives=True
231
  ).execute()
 
 
232
  except HttpError as e:
233
- return f"❌ 取得資料夾資訊失敗(可能 ID 錯誤或沒權限):{e}"
234
-
235
- if folder_meta.get("mimeType") != "application/vnd.google-apps.folder":
236
- return f"⚠️ 這個 ID 不是資料夾:{folder_meta.get('name')} ({folder_meta.get('mimeType')})"
237
 
238
  try:
239
- query = f"'{folder_id}' in parents and trashed=false"
240
  res = service.files().list(
241
- q=query,
242
  fields="files(id, name, mimeType, webViewLink)",
243
  pageSize=50,
244
  supportsAllDrives=True,
@@ -247,31 +234,26 @@ def list_folder_contents(drive_folder_id_input: str):
247
  ).execute()
248
  files = res.get("files", [])
249
  except HttpError as e:
250
- return f"❌ 列出資料夾內容失敗:{e}"
251
 
252
  lines = [
253
  f"✅ Service account:`{client_email}`",
254
- f"✅ 資料夾:**{folder_meta.get('name')}** (`{folder_id}`)",
255
  "📁 內容(最多 50 筆):"
256
  ]
257
  if not files:
258
- lines.append("(空資料夾或沒有讀取權限的檔案)")
259
  else:
260
  for f in files:
261
- lines.append(f"- `{f.get('name')}` | `{f.get('mimeType')}` | {f.get('webViewLink')}")
262
  return "\n".join(lines)
263
 
264
-
265
- # ----------------- Gradio UI -----------------
266
  with gr.Blocks(title="Map → Google Drive Uploader") as demo:
267
  gr.Markdown("""
268
  # 🗺️ Map → Google Drive Uploader
269
- 貼上 `lon,lat`(每行一組),設定視窗範圍並上傳至 Google Drive。
270
- **設定提醒:**
271
- - 到 Space **Settings → Secrets** 新增:
272
- - `SERVICE_ACCOUNT_JSON`:貼完整的 Service Account JSON
273
- - (可選)`DRIVE_FOLDER_ID`:Google Drive 資料夾 ID(或直接貼整個 URL,程式會自動擷取)
274
- - 若要上傳到指定資料夾,請先在 Google Drive 將該資料夾**分享給** Service Account 的 email(編輯者)。
275
  """)
276
 
277
  with gr.Row():
@@ -289,9 +271,9 @@ with gr.Blocks(title="Map → Google Drive Uploader") as demo:
289
  min_lat_in = gr.Number(label="min_lat", value=DEFAULT_EXTENT[2])
290
  max_lat_in = gr.Number(label="max_lat", value=DEFAULT_EXTENT[3])
291
 
292
- folder_id_in = gr.Textbox(
293
- label="Google Drive Folder ID 或整個 URL(可選,覆蓋 Space Secret)",
294
- placeholder="e.g. 1AbCdEfGhIJKlmnopQRstuVWxyz 或 https://drive.google.com/drive/folders/1AbC...",
295
  )
296
  make_public_in = gr.Checkbox(
297
  label="建立公開連結(anyone with the link, reader)", value=False
@@ -311,11 +293,11 @@ with gr.Blocks(title="Map → Google Drive Uploader") as demo:
311
  run_btn.click(
312
  fn=generate_and_upload,
313
  inputs=[points_in, title_in, min_lon_in, max_lon_in, min_lat_in, max_lat_in,
314
- folder_id_in, make_public_in],
315
  outputs=[out_image, out_msg],
316
  )
317
  cfg_btn.click(fn=check_config, inputs=None, outputs=cfg_out)
318
- list_btn.click(fn=list_folder_contents, inputs=folder_id_in, outputs=list_out)
319
 
320
  if __name__ == "__main__":
321
- demo.launch()
 
13
  from googleapiclient.errors import HttpError
14
  from google.oauth2.service_account import Credentials
15
 
16
+ # ====== CONFIG ======
17
  DEFAULT_EXTENT = (119.0, 123.5, 21.5, 25.5) # Taiwan region
18
+ SCOPES = ["https://www.googleapis.com/auth/drive.file"] # create/manage files created by this app
19
+ DEFAULT_FOLDER_URL = "https://drive.google.com/drive/folders/1B7D4Z6vV_G30nIrBr8qFsgNA4l2G-dRo" # 你的指定資料夾
20
+
21
+ # ====== UTILS ======
22
+ def clean_folder_id(folder_input: str) -> Optional[str]:
23
+ """
24
+ 接受整個 Drive 資料夾網址或純 ID,回傳乾淨的 folder ID。
25
+ 會自動移除結尾的 ., 。, 引號與常見標點。
26
+ """
27
+ if not folder_input:
28
+ return None
29
+ raw = folder_input.strip()
30
+
31
+ # 從 URL 擷取 /folders/<ID>
32
+ m = re.search(r"/folders/([A-Za-z0-9_-]+)", raw)
33
+ if m:
34
+ raw = m.group(1)
35
+
36
+ # 移除 query string 與尾端雜訊
37
+ raw = raw.split("?")[0]
38
+ raw = raw.strip().strip('"\'')
39
 
40
+ # 去掉常見尾端符號(半形/全形)
41
+ raw = raw.rstrip(" .。;;,,!!??、/\\』」)》)】]}")
42
 
43
+ # 驗證 ID 合法性
44
+ return raw if re.fullmatch(r"[A-Za-z0-9_-]{20,}", raw) else None
45
+
46
+
47
+ def parse_points(csv_text: str) -> List[Tuple[float, float]]:
48
  pts: List[Tuple[float, float]] = []
49
  if not csv_text:
50
  return pts
 
63
  return pts
64
 
65
 
66
+ def make_figure(points: List[Tuple[float, float]], title: str,
67
+ extent: Tuple[float, float, float, float] = DEFAULT_EXTENT,
68
+ dpi: int = 150) -> str:
 
69
  min_lon, max_lon, min_lat, max_lat = extent
70
  fig, ax = plt.subplots(figsize=(6.5, 6.5), dpi=dpi)
71
  ax.set_title(title or "Map Plot", pad=10)
 
85
  return out_path
86
 
87
 
88
+ def drive_service_from_secret():
 
89
  raw = os.environ.get("SERVICE_ACCOUNT_JSON", "").strip()
90
  if not raw:
91
+ return None, None, "SERVICE_ACCOUNT_JSON not found."
92
  try:
93
  info = json.loads(raw)
94
  except Exception as e:
 
102
  return None, None, f"Failed to build Drive service: {e}"
103
 
104
 
105
+ def upload_png_to_drive(local_png: str, service, folder_id: Optional[str], make_public: bool):
106
+ meta = {"name": os.path.basename(local_png)}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
107
  if folder_id:
108
  meta["parents"] = [folder_id]
109
+ with open(local_png, "rb") as f:
 
 
 
110
  media = MediaIoBaseUpload(f, mimetype="image/png", resumable=False)
111
  created = (service.files()
112
  .create(body=meta, media_body=media,
113
  fields="id, webViewLink, webContentLink",
114
  supportsAllDrives=True)
115
  .execute())
116
+ file_id = created.get("id", "")
117
+ web_view = created.get("webViewLink", "")
118
 
119
+ if make_public and file_id:
120
  try:
121
  service.permissions().create(
122
  fileId=file_id,
 
124
  fields="id",
125
  supportsAllDrives=True,
126
  ).execute()
 
127
  except HttpError as e:
128
+ web_view += f"\n(設定公開連結失敗: {e})"
129
+ return file_id, web_view
130
 
131
+ # ====== GRADIO HANDLERS ======
132
+ def generate_and_upload(points_text, title, min_lon, max_lon, min_lat, max_lat,
133
+ folder_input, make_public):
134
+ # 1) 畫圖
 
 
135
  try:
136
+ pts = parse_points(points_text)
137
+ png = make_figure(pts, title, extent=(min_lon, max_lon, min_lat, max_lat))
 
138
  except Exception as e:
139
  return None, f"❌ Failed to render figure: {e}"
140
 
141
+ # 2) Drive 服務
142
+ service, client_email, err = drive_service_from_secret()
143
  if service is None:
144
+ return png, ("⚠️ 未設定 Google Drive:請在 Settings Secrets 新增 `SERVICE_ACCOUNT_JSON`。\n"
145
+ f"Debug: {err}")
 
 
 
146
 
147
+ # 3) 取得資料夾 ID(優先:UI Secret → 預設網址)
148
+ env_folder = os.environ.get("DRIVE_FOLDER_ID", "").strip()
149
+ raw_folder = folder_input or env_folder or DEFAULT_FOLDER_URL
150
+ folder_id = clean_folder_id(raw_folder)
151
 
152
+ if not folder_id:
153
+ return png, (" 無效的 Folder ID/URL。請貼整個資料夾網址或正確的 ID。\n"
154
+ f"收到:`{raw_folder}`")
155
 
156
+ # 4) 上傳
157
  try:
158
+ file_id, view_link = upload_png_to_drive(png, service, folder_id, make_public)
159
+ msg = [
160
  "✅ Uploaded to Google Drive.",
161
+ f"• Service account: `{client_email}`",
162
+ f"• Folder ID used: `{folder_id}`",
163
  f"• File ID: `{file_id}`",
 
164
  ]
165
  if view_link:
166
+ msg.append(f"• Open: {view_link}")
167
  if not make_public:
168
+ msg.append("• Note: 未勾選公開,檔案權限沿用資料夾/硬碟設定。")
169
+ return png, "\n".join(msg)
170
  except HttpError as e:
171
+ return png, f"❌ Upload failed (HttpError): {e}"
172
  except Exception as e:
173
+ return png, f"❌ Upload failed: {e}"
174
 
175
 
176
  def check_config():
 
177
  raw = os.environ.get("SERVICE_ACCOUNT_JSON", None)
178
+ env_folder = os.environ.get("DRIVE_FOLDER_ID", "").strip()
179
+ default_used = DEFAULT_FOLDER_URL
180
+ folder_id = clean_folder_id(env_folder or default_used)
181
 
182
  if raw is None:
183
+ return "❌ `SERVICE_ACCOUNT_JSON` 不存在。"
 
 
184
  try:
185
  info = json.loads(raw)
186
+ client_email = info.get("client_email")
187
  except Exception as e:
188
  return f"❌ `SERVICE_ACCOUNT_JSON` 不是有效 JSON:{e}"
 
189
 
190
+ out = [f"✅ 讀到 SERVICE_ACCOUNT_JSON,client_email = {client_email}",
191
+ f"• Folder ID (Secret/Default):`{folder_id}`"]
192
  try:
193
+ service, _, err = drive_service_from_secret()
194
  if service and folder_id:
195
  meta = service.files().get(
196
  fileId=folder_id, fields="id, name", supportsAllDrives=True
197
  ).execute()
198
+ out.append(f"• Folder name: {meta.get('name')}")
199
  elif err:
200
+ out.append(f"service 建立失敗:{err}")
201
  except HttpError as e:
202
+ out.append(f" 讀取資料夾資訊失敗:{e}")
203
+ return "\n".join(out)
 
204
 
205
+ def list_folder(folder_input):
206
+ service, client_email, err = drive_service_from_secret()
 
 
207
  if service is None:
208
+ return f"❌ 無法建立 Drive 服務:{err}"
209
 
210
+ env_folder = os.environ.get("DRIVE_FOLDER_ID", "").strip()
211
+ raw_folder = folder_input or env_folder or DEFAULT_FOLDER_URL
212
+ folder_id = clean_folder_id(raw_folder)
213
  if not folder_id:
214
+ return "❌ 無效的 Folder ID/URL。"
215
 
216
  try:
217
+ meta = service.files().get(
218
  fileId=folder_id, fields="id, name, mimeType",
219
  supportsAllDrives=True
220
  ).execute()
221
+ if meta.get("mimeType") != "application/vnd.google-apps.folder":
222
+ return f"⚠️ 這不是資料夾:{meta.get('name')} ({meta.get('mimeType')})"
223
  except HttpError as e:
224
+ return f"❌ 取得資料夾資訊失敗:{e}"
 
 
 
225
 
226
  try:
 
227
  res = service.files().list(
228
+ q=f"'{folder_id}' in parents and trashed=false",
229
  fields="files(id, name, mimeType, webViewLink)",
230
  pageSize=50,
231
  supportsAllDrives=True,
 
234
  ).execute()
235
  files = res.get("files", [])
236
  except HttpError as e:
237
+ return f"❌ 列出內容失敗:{e}"
238
 
239
  lines = [
240
  f"✅ Service account:`{client_email}`",
241
+ f"✅ 資料夾:**{meta.get('name')}** (`{folder_id}`)",
242
  "📁 內容(最多 50 筆):"
243
  ]
244
  if not files:
245
+ lines.append("(空資料夾或無可見檔案)")
246
  else:
247
  for f in files:
248
+ lines.append(f"- `{f['name']}` | `{f['mimeType']}` | {f['webViewLink']}")
249
  return "\n".join(lines)
250
 
251
+ # ====== UI ======
 
252
  with gr.Blocks(title="Map → Google Drive Uploader") as demo:
253
  gr.Markdown("""
254
  # 🗺️ Map → Google Drive Uploader
255
+ 預設會上傳到:`https://drive.google.com/drive/folders/1B7D4Z6vV_G30nIrBr8qFsgNA4l2G-dRo`
256
+ (也可在下方覆蓋資料夾 URL/ID;或在 Secrets 設 `DRIVE_FOLDER_ID`)
 
 
 
 
257
  """)
258
 
259
  with gr.Row():
 
271
  min_lat_in = gr.Number(label="min_lat", value=DEFAULT_EXTENT[2])
272
  max_lat_in = gr.Number(label="max_lat", value=DEFAULT_EXTENT[3])
273
 
274
+ folder_in = gr.Textbox(
275
+ label="(可選)覆蓋資料夾:貼整個 URL 或純 ID",
276
+ placeholder=DEFAULT_FOLDER_URL,
277
  )
278
  make_public_in = gr.Checkbox(
279
  label="建立公開連結(anyone with the link, reader)", value=False
 
293
  run_btn.click(
294
  fn=generate_and_upload,
295
  inputs=[points_in, title_in, min_lon_in, max_lon_in, min_lat_in, max_lat_in,
296
+ folder_in, make_public_in],
297
  outputs=[out_image, out_msg],
298
  )
299
  cfg_btn.click(fn=check_config, inputs=None, outputs=cfg_out)
300
+ list_btn.click(fn=list_folder, inputs=folder_in, outputs=list_out)
301
 
302
  if __name__ == "__main__":
303
+ demo.launch()