mautrix-instagram/mauigpapi/http/account.py

# mautrix-instagram - A Matrix-Instagram puppeting bridge.
# Copyright (C) 2020 Tulir Asokan
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
from typing import Optional, Type, TypeVar
import json

from ..types import CurrentUserResponse
from .base import BaseAndroidAPI

T = TypeVar('T')


class AccountAPI(BaseAndroidAPI):
    async def current_user(self) -> CurrentUserResponse:
        return await self.std_http_get(f"/api/v1/accounts/current_user/", query={"edit": "true"},
                                       response_type=CurrentUserResponse)

    async def set_biography(self, text: str) -> CurrentUserResponse:
        # TODO entities?
        return await self.__command("set_biography", device_id=self.state.device.id, raw_text=text)

    async def set_profile_picture(self, upload_id: str) -> CurrentUserResponse:
        return await self.__command("change_profile_picture",
                                    use_fbuploader="true", upload_id=upload_id)

    async def remove_profile_picture(self) -> CurrentUserResponse:
        return await self.__command("remove_profile_picture")

    async def set_private(self, private: bool) -> CurrentUserResponse:
        return await self.__command("set_private" if private else "set_public")

    async def confirm_email(self, slug: str) -> CurrentUserResponse:
        # slug can contain slashes, but it shouldn't start or end with one
        return await self.__command(f"confirm_email/{slug}")

    async def send_recovery_flow_email(self, query: str):
        req = {
            "_csrftoken": self.state.cookies.csrf_token,
            "adid": "",
            "guid": self.state.device.uuid,
            "device_id": self.state.device.id,
            "query": query,
        }
        # TODO parse response content
        return await self.std_http_post(f"/api/v1/accounts/send_recovery_flow_email/", data=req)

    async def edit_profile(self, external_url: Optional[str] = None, gender: Optional[str] = None,
                           phone_number: Optional[str] = None, username: Optional[str] = None,
                           # TODO should there be a last_name?
                           first_name: Optional[str] = None, biography: Optional[str] = None,
                           email: Optional[str] = None) -> CurrentUserResponse:
        return await self.__command("edit_profile", device_id=self.state.device.id, email=email,
                                    external_url=external_url, first_name=first_name,
                                    username=username, phone_number=phone_number, gender=gender,
                                    biography=biography)

    async def __command(self, command: str, response_type: Type[T] = CurrentUserResponse,
                        **kwargs: str) -> T:
        req = {
            "_csrftoken": self.state.cookies.csrf_token,
            "_uid": self.state.cookies.user_id,
            "_uuid": self.state.device.uuid,
            **kwargs,
        }
        return await self.std_http_post(f"/api/v1/accounts/{command}/", data=req,
                                        filter_nulls=True, response_type=response_type)

    async def read_msisdn_header(self, usage: str = "default"):
        req = {
            "mobile_subno_usage": usage,
            "device_id": self.state.device.uuid,
        }
        headers = {
            "X-DEVICE-ID": self.state.device.uuid,
        }
        return await self.std_http_post("/api/v1/accounts/read_msisdn_header/", data=req,
                                        headers=headers)

    async def msisdn_header_bootstrap(self, usage: str = "default"):
        req = {
            "mobile_subno_usage": usage,
            "device_id": self.state.device.uuid,
        }
        return await self.std_http_post("/api/v1/accounts/msisdn_header_bootstrap/", data=req)

    async def contact_point_prefill(self, usage: str = "default"):
        req = {
            "mobile_subno_usage": usage,
            "device_id": self.state.device.uuid,
        }
        return await self.std_http_post("/api/v1/accounts/contact_point_prefill/", data=req)

    async def get_prefill_candidates(self):
        req = {
            "android_device_id": self.state.device.id,
            "usages": json.dumps(["account_recovery_omnibox"]),
            "device_id": self.state.device.uuid,
        }
        # TODO parse response content
        return await self.std_http_post("/api/v1/accounts/get_prefill_candidates/", data=req)

    async def process_contact_point_signals(self):
        req = {
            "phone_id": self.state.device.phone_id,
            "_csrftoken": self.state.cookies.csrf_token,
            "_uid": self.state.cookies.user_id,
            "device_id": self.state.device.uuid,
            "_uuid": self.state.device.uuid,
            "google_tokens": json.dumps([]),
        }
        return await self.std_http_post("/api/v1/accounts/process_contact_point_signals/",
                                        data=req)