import os import json from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from event_logger import get_logger class GoogleDriveUploader: def __init__(self, config_path='./config.json'): with open(config_path, 'r') as f: self.config = json.load(f) self.folder_id = self.config['google_drive_folder_id'] self.creds_path = self.config['google_credentials_path'] self.scopes = ['https://www.googleapis.com/auth/drive.file'] self.service = None self.authenticate() def authenticate(self): """ Google Drive API 인증 """ # 자격증명 파일 존재 여부 확인 (명시적으로 친절한 메시지 제공) if not self.creds_path or not os.path.isfile(self.creds_path): raise FileNotFoundError( f"Google Drive API 자격증명 파일을 찾을 수 없습니다: {self.creds_path}.\n" f"config.json의 'google_credentials_path'를 올바른 credentials.json 경로로 설정하거나, API 업로드를 사용하지 마세요." ) token_path = os.path.join(os.path.dirname(self.creds_path) or '.', 'token.json') creds = None if os.path.exists(token_path): creds = Credentials.from_authorized_user_file(token_path, self.scopes) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( self.creds_path, self.scopes) creds = flow.run_local_server(port=0) with open(token_path, 'w') as token: token.write(creds.to_json()) self.service = build('drive', 'v3', credentials=creds) def upload_file(self, file_path, file_name=None): """ 파일을 Google Drive에 업로드 """ logger = get_logger() if self.service is None: raise RuntimeError('Google Drive API가 초기화되지 않았습니다. credentials.json과 folder_id를 설정하세요.') if file_name is None: file_name = os.path.basename(file_path) file_metadata = { 'name': file_name, 'parents': [self.folder_id] if self.folder_id else [] } media = MediaFileUpload(file_path, resumable=True) try: file = self.service.files().create( body=file_metadata, media_body=media, fields='id' ).execute() fid = file.get('id') print(f'파일 업로드 완료: {file_name} (ID: {fid})') if logger: logger.log_event("drive_upload", name=file_name, id=fid) return fid except Exception as e: print(f'업로드 실패: {e}') if logger: logger.log_event("drive_upload_error", name=file_name, error=str(e)) return None def upload_data_as_json(self, data, filename='collected_data.json'): """ 데이터를 JSON 파일로 변환하여 업로드 """ import tempfile logger = get_logger() with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: json.dump(data, f, ensure_ascii=False, indent=2) temp_path = f.name try: file_id = self.upload_file(temp_path, filename) logger = get_logger() if logger and logger.preview_saved_files: try: with open(temp_path, 'r', encoding='utf-8') as rf: content = rf.read(logger.preview_limit) logger.log_event("file_preview", name=filename, preview=content) except Exception: pass return file_id finally: os.unlink(temp_path) if logger: logger.log_event("tempfile_cleanup", path=temp_path) def list_files(self): """ 폴더 내 파일 목록 조회 """ if self.service is None: raise RuntimeError('Google Drive API가 초기화되지 않았습니다. credentials.json과 folder_id를 설정하세요.') try: results = self.service.files().list( q=f"'{self.folder_id}' in parents", pageSize=10, fields="nextPageToken, files(id, name)" ).execute() items = results.get('files', []) logger = get_logger() if logger: logger.log_event("drive_list", count=len(items)) return items except Exception as e: print(f'파일 목록 조회 실패: {e}') logger = get_logger() if logger: logger.log_event("drive_list_error", error=str(e)) return [] class SimpleDriveSaver: """ Colab의 drive.mount()를 사용한 간단한 저장 방식 """ def __init__(self, mount_path='/content/drive/MyDrive/AI_Data'): self.mount_path = mount_path if not os.path.exists(mount_path): os.makedirs(mount_path, exist_ok=True) def save_data_as_json(self, data, filename='collected_data.json'): """ 데이터를 마운트된 Drive에 JSON 파일로 저장 """ filepath = os.path.join(self.mount_path, filename) try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f'데이터 저장 완료: {filepath}') logger = get_logger() if logger: logger.log_event("file_saved", path=filepath, bytes=os.path.getsize(filepath)) if logger.preview_saved_files: try: with open(filepath, 'r', encoding='utf-8') as rf: preview = rf.read(logger.preview_limit) logger.log_event("file_preview", path=filepath, preview=preview) except Exception: pass return filepath except Exception as e: print(f'저장 실패: {e}') logger = get_logger() if logger: logger.log_event("file_save_error", path=filepath, error=str(e)) return None def save_text_data(self, data, filename='collected_data.txt'): """ 데이터를 텍스트 파일로 저장 """ filepath = os.path.join(self.mount_path, filename) try: with open(filepath, 'w', encoding='utf-8') as f: if isinstance(data, list): for item in data: f.write(json.dumps(item, ensure_ascii=False) + '\n') else: f.write(str(data)) print(f'텍스트 데이터 저장 완료: {filepath}') logger = get_logger() if logger: logger.log_event("file_saved", path=filepath, bytes=os.path.getsize(filepath)) return filepath except Exception as e: print(f'저장 실패: {e}') logger = get_logger() if logger: logger.log_event("file_save_error", path=filepath, error=str(e)) return None def save_to_drive_simple(data, filename='collected_data.json', mount_path='/content/drive/MyDrive/AI_Data'): """ 간단한 함수로 마운트된 Drive에 데이터 저장 """ saver = SimpleDriveSaver(mount_path) return saver.save_data_as_json(data, filename)