phone-number-migration/Phone_number_migration.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yLL1WjbsNJdP"
      },
      "source": [
        "# Phone number migration\n",
        "\n",
        "**Перед запуском необходимо добавить файл с Oauth2 Сredentials (по умолчанию credentials.json)**\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WmCKpRemMRba"
      },
      "source": [
        "Импортируем необходимые библиотеки."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "pzHzXS_eFteU"
      },
      "outputs": [],
      "source": [
        "import os.path\n",
        "\n",
        "from google.auth.transport.requests import Request\n",
        "from google.oauth2.credentials import Credentials\n",
        "from google_auth_oauthlib.flow import InstalledAppFlow\n",
        "from googleapiclient.discovery import build\n",
        "from googleapiclient.errors import HttpError"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qA-EnsZIMOeZ"
      },
      "source": [
        "Основные параметры приложения."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "id": "rRgABs8rF1d8"
      },
      "outputs": [],
      "source": [
        "# Параметры приложения\n",
        "OLD_NUMBER_PREFIXES = ['+38071', '071']\n",
        "NEW_NUMBER_PREFIXES = ['+7949']\n",
        "\n",
        "# Функция для получения нового номера из основы старого\n",
        "# Вернет номер в формате +7 949 123-45-67\n",
        "def new_phone_number(base: str):\n",
        "    num = '7949' + base\n",
        "    return '+{} {} {}-{}-{}'.format(\n",
        "        num[0],\n",
        "        num[1:4],\n",
        "        num[4:7],\n",
        "        num[7:9],\n",
        "        num[9:11]\n",
        "    )\n",
        "# Тип телефона.\n",
        "# Доступные значения: https://developers.google.com/people/api/rest/v1/people#phonenumber\n",
        "PHONE_TYPE = 'mobile'"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "YMHU4Hs_F5-Y"
      },
      "outputs": [],
      "source": [
        "# Параметры Google API\n",
        "CREDENTIALS_FILE = 'credentials.json'\n",
        "TOKEN_FILE = 'token.json'\n",
        "SCOPES = ['https://www.googleapis.com/auth/contacts']"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "Функция для получения доступа к Google аккаунту:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {},
      "outputs": [],
      "source": [
        "def google_creds():\n",
        "    creds = None\n",
        "\n",
        "    if os.path.exists(TOKEN_FILE):\n",
        "        creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)\n",
        "\n",
        "    if not creds or not creds.valid:\n",
        "        if creds and creds.expired and creds.refresh_token:\n",
        "            creds.refresh(Request())\n",
        "        else:\n",
        "            flow = InstalledAppFlow.from_client_secrets_file(\n",
        "                CREDENTIALS_FILE, SCOPES)\n",
        "            creds = flow.run_console()\n",
        "            with open(TOKEN_FILE, 'w') as token:\n",
        "                token.write(creds.to_json())\n",
        "\n",
        "    return creds"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-Kdstr4TOs1H"
      },
      "source": [
        "Основной код приложения:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "TimQdg09F2GV"
      },
      "outputs": [],
      "source": [
        "def main():\n",
        "    creds = None\n",
        "    if os.path.exists(TOKEN_FILE):\n",
        "        creds = Credentials.from_authorized_user_file(TOKEN_FILE, SCOPES)\n",
        "    if not creds or not creds.valid:\n",
        "        if creds and creds.expired and creds.refresh_token:\n",
        "            creds.refresh(Request())\n",
        "        else:\n",
        "            flow = InstalledAppFlow.from_client_secrets_file(\n",
        "                CREDENTIALS_FILE, SCOPES)\n",
        "            creds = flow.run_console()\n",
        "            with open(TOKEN_FILE, 'w') as token:\n",
        "                token.write(creds.to_json())\n",
        "    try:\n",
        "        service = build('people', 'v1', credentials=creds)\n",
        "        print('Получаем список контактов...')\n",
        "        results = service.people().connections().list(resourceName='people/me',\n",
        "                                                      personFields='names,phoneNumbers').execute()\n",
        "        connections = results.get('connections', [])\n",
        "        print('Количество контактов: {}'.format(len(connections)))\n",
        "        batch_request = {\n",
        "            \"contacts\": {},\n",
        "            \"updateMask\": \"phoneNumbers\"\n",
        "        }\n",
        "        for person in connections:\n",
        "            phoneNumbers = person.get('phoneNumbers', [])\n",
        "            has_old = False\n",
        "            old_numbers = set()\n",
        "            new_numbers = set()\n",
        "            # Получаем номера нового и старого формата\n",
        "            for phoneNumber in phoneNumbers:\n",
        "                num = phoneNumber.get('canonicalForm')\n",
        "                if not num:\n",
        "                    continue\n",
        "                base = get_number_without_prefix(num, OLD_NUMBER_PREFIXES)\n",
        "                if (base):\n",
        "                    has_old = True\n",
        "                    old_numbers.add(base)\n",
        "                base = get_number_without_prefix(num, NEW_NUMBER_PREFIXES)\n",
        "                if (base):\n",
        "                    new_numbers.add(base)\n",
        "            # Если нет старого - пропускаем\n",
        "            if not has_old:\n",
        "                continue\n",
        "            # Получаем старые номера, у которых нету парного нового\n",
        "            numbers_not_updated = list(\n",
        "                old_numbers.difference(new_numbers)\n",
        "            )\n",
        "        \n",
        "            # Если нет старых номеров, которые нужно обновить - пропускаем\n",
        "            if len(numbers_not_updated) == 0:\n",
        "                continue\n",
        "            # Получаем новые номера\n",
        "            new_phone_numbers = [\n",
        "                new_phone_number(x) for x in numbers_not_updated\n",
        "            ]\n",
        "            # batchUpdateContacts: https://developers.google.com/people/api/rest/v1/people/batchUpdateContacts\n",
        "            # PhoneNumber: https://developers.google.com/people/api/rest/v1/people#phonenumber\n",
        "            batch_request['contacts'][person['resourceName']] = {\n",
        "                \"phoneNumbers\": [\n",
        "                    *person['phoneNumbers'],\n",
        "                    [\n",
        "                        { \n",
        "                            \"value\": number, \n",
        "                            \"type\": PHONE_TYPE\n",
        "                        } for number in new_phone_numbers\n",
        "                    ]\n",
        "            ],\n",
        "            \"etag\": person['etag']\n",
        "        }\n",
        "        contacts_len = len(batch_request['contacts'])\n",
        "        if (contacts_len > 0):\n",
        "            print('Количество контактов для обновления: {}'.format(contacts_len))\n",
        "            print('Обновляем контакты...')\n",
        "            service.people().batchUpdateContacts(body = batch_request).execute()\n",
        "            print('Обновление завершено!')\n",
        "        else:\n",
        "            print('Нечего обновлять!')\n",
        "    except HttpError as err:\n",
        "        print(err)\n",
        "\n",
        "def get_number_without_prefix(number: str, prefixes: list):\n",
        "    for prefix in prefixes:\n",
        "        if number.startswith(prefix):\n",
        "            return number[len(prefix):]\n",
        "    return False"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WnbwQVZPNDpP"
      },
      "source": [
        "Запуск приложения:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 332
        },
        "id": "0v8m0ZYmGHnC",
        "outputId": "092f3a86-80d3-45c7-d4d6-f72878d96017"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Получаем список контактов...\n",
            "Количество контактов: 48\n",
            "Нечего обновлять!\n"
          ]
        }
      ],
      "source": [
        "if __name__ == '__main__':\n",
        "    main()\n"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "name": "Phone number migration.ipynb",
      "provenance": []
    },
    "interpreter": {
      "hash": "e7370f93d1d0cde622a1f8e1c04877d8463912d04d973331ad4851f04de6915a"
    },
    "kernelspec": {
      "display_name": "Python 3.10.4 64-bit",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "codemirror_mode": {
        "name": "ipython",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.10.4"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}