#!/usr/bin/env python3 """ GeoNames to Google Sheets - Complete Implementation ================================================== Kompletní verze s funkčním Google Sheets API přístupem. """ import requests import csv import logging from datetime import datetime from typing import List, Dict, Optional import json import os # Google API imports try: from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build GOOGLE_AVAILABLE = True except ImportError: GOOGLE_AVAILABLE = False print("⚠️ Google API libraries not installed. Run: pip install -r requirements.txt") # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('geonames_sync.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) SCOPES = ['https://www.googleapis.com/auth/spreadsheets'] class GeoNamesProcessor: """Vylepšená verze pro stahování a zpracování GeoNames dat""" def __init__(self): self.country_info_url = "https://download.geonames.org/export/dump/countryInfo.txt" self.data = [] def download_and_process(self) -> List[Dict]: """Stáhni a zpracuj data z GeoNames""" try: logger.info("Downloading country data from GeoNames...") response = requests.get(self.country_info_url, timeout=30) response.raise_for_status() lines = response.text.strip().split('\n') processed_data = [] for line in lines: if line.startswith('#') or not line.strip(): continue fields = line.split('\t') if len(fields) >= 17: # Čištění a formátování dat country_data = { 'ISO Alpha2': fields[0].strip(), 'ISO Alpha3': fields[1].strip(), 'ISO Numeric': fields[2].strip(), 'FIPS Code': fields[3].strip(), 'Country Name': fields[4].strip(), 'Capital': fields[5].strip(), 'Area (km²)': self._format_number(fields[6]), 'Population': self._format_number(fields[7]), 'Continent': self._format_continent(fields[8]), 'Domain': fields[9].strip(), 'Currency Code': fields[10].strip(), 'Currency Name': fields[11].strip(), 'Phone Code': fields[12].strip(), 'Postal Format': fields[13].strip(), 'Languages': fields[15].strip() if len(fields) > 15 else '', 'GeoName ID': fields[16].strip() if len(fields) > 16 else '', 'Neighbours': fields[17].strip() if len(fields) > 17 else '', 'Last Updated': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } processed_data.append(country_data) # Seřazení podle názvu země processed_data.sort(key=lambda x: x['Country Name']) logger.info(f"Successfully processed {len(processed_data)} countries") self.data = processed_data return processed_data except Exception as e: logger.error(f"Error downloading/processing data: {e}") raise def _format_number(self, value: str) -> str: """Formátuj čísla pro lepší čitelnost""" try: if value and value.strip() and value.strip() != '0': num = int(float(value.strip())) return f"{num:,}".replace(',', ' ') # Použij mezery místo čárek return value.strip() except: return value.strip() def _format_continent(self, code: str) -> str: """Převeď kódy kontinentů na jména""" continent_map = { 'AF': 'Africa', 'AS': 'Asia', 'EU': 'Europe', 'NA': 'North America', 'SA': 'South America', 'OC': 'Oceania', 'AN': 'Antarctica' } return continent_map.get(code.strip(), code.strip()) class GoogleSheetsManager: """Kompletní správa Google Sheets API""" def __init__(self, spreadsheet_id: str): self.spreadsheet_id = spreadsheet_id self.service = None self._setup_credentials() def _setup_credentials(self): """Nastav Google API credentials""" if not GOOGLE_AVAILABLE: logger.error("Google API libraries not available") return False creds = None # Load existing token if os.path.exists('token.json'): creds = Credentials.from_authorized_user_file('token.json', SCOPES) # If no valid credentials, get new ones if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: logger.info("Refreshing credentials...") creds.refresh(Request()) else: if not os.path.exists('credentials.json'): logger.error("credentials.json not found! Please run setup_google_api.py first") return False logger.info("Starting OAuth flow...") flow = InstalledAppFlow.from_client_secrets_file('credentials.json', SCOPES) creds = flow.run_local_server(port=0) # Save credentials with open('token.json', 'w') as token: token.write(creds.to_json()) try: self.service = build('sheets', 'v4', credentials=creds) logger.info("Google Sheets API connection established") return True except Exception as e: logger.error(f"Failed to build service: {e}") return False def clear_sheet(self, sheet_name: str = "Sheet1") -> bool: """Vymaž obsah listu""" try: range_name = f"{sheet_name}!A:Z" self.service.spreadsheets().values().clear( spreadsheetId=self.spreadsheet_id, range=range_name ).execute() logger.info(f"Cleared sheet: {sheet_name}") return True except Exception as e: logger.error(f"Error clearing sheet: {e}") return False def upload_data(self, data: List[Dict], sheet_name: str = "Sheet1", clear_first: bool = True) -> bool: """Nahraj data do Google Sheets""" try: if not self.service: logger.error("Google Sheets service not available") return False if not data: logger.warning("No data to upload") return False # Clear sheet first if requested if clear_first: self.clear_sheet(sheet_name) # Prepare data for upload headers = list(data[0].keys()) values = [headers] # Header row # Add data rows for row in data: values.append([str(row.get(header, '')) for header in headers]) # Upload to Google Sheets body = { 'values': values, 'majorDimension': 'ROWS' } result = self.service.spreadsheets().values().update( spreadsheetId=self.spreadsheet_id, range=f"{sheet_name}!A1", valueInputOption="USER_ENTERED", # Umožní formátování čísel body=body ).execute() updated_cells = result.get('updatedCells', 0) logger.info(f"Successfully uploaded {len(values)} rows ({updated_cells} cells) to {sheet_name}") # Format the sheet self._format_sheet(sheet_name, len(headers)) return True except Exception as e: logger.error(f"Error uploading data: {e}") return False def _format_sheet(self, sheet_name: str, num_columns: int): """Formátuj sheet pro lepší čitelnost""" try: # Get sheet ID spreadsheet = self.service.spreadsheets().get(spreadsheetId=self.spreadsheet_id).execute() sheet_id = None for sheet in spreadsheet.get('sheets', []): if sheet.get('properties', {}).get('title') == sheet_name: sheet_id = sheet.get('properties', {}).get('sheetId') break if sheet_id is None: logger.warning(f"Could not find sheet ID for {sheet_name}") return # Format requests requests = [ # Freeze header row { "updateSheetProperties": { "properties": { "sheetId": sheet_id, "gridProperties": { "frozenRowCount": 1 } }, "fields": "gridProperties.frozenRowCount" } }, # Bold header row { "repeatCell": { "range": { "sheetId": sheet_id, "startRowIndex": 0, "endRowIndex": 1, "startColumnIndex": 0, "endColumnIndex": num_columns }, "cell": { "userEnteredFormat": { "textFormat": { "bold": True }, "backgroundColor": { "red": 0.9, "green": 0.9, "blue": 0.9 } } }, "fields": "userEnteredFormat(textFormat,backgroundColor)" } }, # Auto-resize columns { "autoResizeDimensions": { "dimensions": { "sheetId": sheet_id, "dimension": "COLUMNS", "startIndex": 0, "endIndex": num_columns } } } ] body = {'requests': requests} self.service.spreadsheets().batchUpdate( spreadsheetId=self.spreadsheet_id, body=body ).execute() logger.info(f"Applied formatting to {sheet_name}") except Exception as e: logger.warning(f"Could not apply formatting: {e}") def main(): """Hlavní funkce""" logger.info("=== GeoNames to Google Sheets Sync Started ===") # Configuration SPREADSHEET_ID = "1b4GKYaRVh_V3dN4Ok01EiUHdpoQxr0a_6ptoSZWsj_o" SHEET_NAME = "Countries Data" # Můžeš změnit název listu try: # Step 1: Download and process data processor = GeoNamesProcessor() countries = processor.download_and_process() if not countries: logger.error("No country data downloaded") return False # Step 2: Save backup CSV backup_file = f"countries_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" with open(backup_file, 'w', newline='', encoding='utf-8') as csvfile: if countries: writer = csv.DictWriter(csvfile, fieldnames=countries[0].keys()) writer.writeheader() writer.writerows(countries) logger.info(f"Backup saved: {backup_file}") # Step 3: Upload to Google Sheets sheets_manager = GoogleSheetsManager(SPREADSHEET_ID) success = sheets_manager.upload_data(countries, SHEET_NAME) if success: logger.info("=== Synchronization completed successfully! ===") print(f"\n✅ Success! Check your Google Sheets:") print(f"📊 Spreadsheet: https://docs.google.com/spreadsheets/d/{SPREADSHEET_ID}/") print(f"📋 Sheet: {SHEET_NAME}") print(f"📈 Countries uploaded: {len(countries)}") return True else: logger.error("=== Synchronization failed ===") return False except Exception as e: logger.error(f"Main execution failed: {e}") return False if __name__ == "__main__": success = main() exit(0 if success else 1)