text: import subprocess
  import shlex
  import json
  import random
  import time
  from pathlib import Path
  from typing import Dict, List
  from uuid import uuid4
  
  from instagrapi import config
  from instagrapi.exceptions import ClientError, IGTVConfigureError, IGTVNotUpload
  from instagrapi.extractors import extract_media_v1
  from instagrapi.types import Location, Media, Usertag
  from instagrapi.utils import date_time_original
  
  try:
      from PIL import Image
  except ImportError:
      raise Exception("You don't have PIL installed. Please install PIL or Pillow>=8.1.1")
  
  
  class DownloadIGTVMixin:
      """
      Helpers to download IGTV videos
      """
  
      def igtv_download(self, media_pk: int, folder: Path = "") -> str:
          """
          Download IGTV video
  
          Parameters
          ----------
          media_pk: int
              PK for the album you want to download
          folder: Path, optional
              Directory in which you want to download the album, default is "" and will download the files to working
                  directory.
  
          Returns
          -------
          str
          """
          return self.video_download(media_pk, folder)
  
      def igtv_download_by_url(
          self, url: str, filename: str = "", folder: Path = ""
      ) -> str:
          """
          Download IGTV video using URL
  
          Parameters
          ----------
          url: str
              URL to download media from
          folder: Path, optional
              Directory in which you want to download the album, default is "" and will download the files to working
                  directory.
  
          Returns
          -------
          str
          """
          return self.video_download_by_url(url, filename, folder)
  
  
  class UploadIGTVMixin:
      """
      Helpers to upload IGTV videos
      """
  
      def igtv_upload(
          self,
          path: Path,
          title: str,
          caption: str,
          thumbnail: Path = None,
          usertags: List[Usertag] = [],
          location: Location = None,
          configure_timeout: int = 10,
          extra_data: Dict[str, str] = {},
      ) -> Media:
          """
          Upload IGTV to Instagram
  
          Parameters
          ----------
          path: Path
              Path to IGTV file
          title: str
              Title of the video
          caption: str
              Media caption
          thumbnail: Path, optional
              Path to thumbnail for IGTV. Default value is None, and it generates a thumbnail
          usertags: List[Usertag], optional
              List of users to be tagged on this upload, default is empty list.
          location: Location, optional
              Location tag for this upload, default is none
          configure_timeout: int
              Timeout between attempt to configure media (set caption, etc), default is 10
          extra_data: Dict[str, str], optional
              Dict of extra data, if you need to add your params, like {"share_to_facebook": 1}.
  
          Returns
          -------
          Media
              An object of Media class
          """
          path = Path(path)
          if thumbnail is not None:
              thumbnail = Path(thumbnail)
          upload_id = str(int(time.time() * 1000))
          thumbnail, width, height, duration = analyze_video(path, thumbnail)
          waterfall_id = str(uuid4())
          # upload_name example: '1576102477530_0_7823256191'
          upload_name = "{upload_id}_0_{rand}".format(
              upload_id=upload_id, rand=random.randint(1000000000, 9999999999)
          )
          # by segments bb2c1d0c127384453a2122e79e4c9a85-0-6498763
          # upload_name = "{hash}-0-{rand}".format(
          #     hash="bb2c1d0c127384453a2122e79e4c9a85", rand=random.randint(1111111, 9999999)
          # )
          rupload_params = {
              "is_igtv_video": "1",
              "retry_context": '{"num_step_auto_retry":0,"num_reupload":0,"num_step_manual_retry":0}',
              "media_type": "2",
              "xsharing_user_ids": json.dumps([self.user_id]),
              "upload_id": upload_id,
              "upload_media_duration_ms": str(int(duration * 1000)),
              "upload_media_width": str(width),
              "upload_media_height": str(height),
          }
          headers = {
              "Accept-Encoding": "gzip",
              "X-Instagram-Rupload-Params": json.dumps(rupload_params),
              "X_FB_VIDEO_WATERFALL_ID": waterfall_id,
              "X-Entity-Type": "video/mp4",
          }
          response = self.private.get(
              "https://{domain}/rupload_igvideo/{name}".format(
                  domain=config.API_DOMAIN, name=upload_name
              ),
              headers=headers,
          )
          self.request_log(response)
          if response.status_code != 200:
              raise IGTVNotUpload(response=self.last_response, **self.last_json)
          with open(path, "rb") as fp:
              igtv_data = fp.read()
              igtv_len = str(len(igtv_data))
          headers = {
              "Offset": "0",
              "X-Entity-Name": upload_name,
              "X-Entity-Length": igtv_len,
              "Content-Type": "application/octet-stream",
              "Content-Length": igtv_len,
              **headers,
          }
          response = self.private.post(
              "https://{domain}/rupload_igvideo/{name}".format(
                  domain=config.API_DOMAIN, name=upload_name
              ),
              data=igtv_data,
              headers=headers,
          )
          self.request_log(response)
          if response.status_code != 200:
              raise IGTVNotUpload(response=self.last_response, **self.last_json)
          # CONFIGURE
          self.igtv_composer_session_id = self.generate_uuid()
          for attempt in range(50):
              self.logger.debug(f"Attempt #{attempt} to configure IGTV: {path}")
              time.sleep(configure_timeout)
              try:
                  configured = self.igtv_configure(
                      upload_id,
                      thumbnail,
                      width,
                      height,
                      duration,
                      title,
                      caption,
                      usertags,
                      location,
                      extra_data=extra_data
                  )
              except ClientError as e:
                  if "Transcode not finished yet" in str(e):
                      """
                      Response 202 status:
                      {"message": "Transcode not finished yet.", "status": "fail"}
                      """
                      time.sleep(configure_timeout)
                      continue
                  raise e
              else:
                  if configured:
                      media = self.last_json.get("media")
                      self.expose()
                      return extract_media_v1(media)
          raise IGTVConfigureError(response=self.last_response, **self.last_json)
  
      def igtv_configure(
          self,
          upload_id: str,
          thumbnail: Path,
          width: int,
          height: int,
          duration: int,
          title: str,
          caption: str,
          usertags: List[Usertag] = [],
          location: Location = None,
          extra_data: Dict[str, str] = {},
      ) -> Dict:
          """
          Post Configure IGTV (send caption, thumbnail and more to Instagram)
  
          Parameters
          ----------
          upload_id: str
              Unique identifier for a IGTV video
          thumbnail: Path
              Path to thumbnail for IGTV
          width: int
              Width of the video in pixels
          height: int
              Height of the video in pixels
          duration: int
              Duration of the video in seconds
          title: str
              Title of the video
          caption: str
              Media caption
          usertags: List[Usertag], optional
              List of users to be tagged on this upload, default is empty list.
          location: Location, optional
              Location tag for this upload, default is None
          extra_data: Dict[str, str], optional
              Dict of extra data, if you need to add your params, like {"share_to_facebook": 1}.
  
          Returns
          -------
          Dict
              A dictionary of response from the call
          """
          self.photo_rupload(Path(thumbnail), upload_id)
          usertags = [
              {"user_id": tag.user.pk, "position": [tag.x, tag.y]} for tag in usertags
          ]
          data = {
              "igtv_ads_toggled_on": "0",
              "filter_type": "0",
              "timezone_offset": str(self.timezone_offset),
              "media_folder": "ScreenRecorder",
              "location": self.location_build(location),
              "source_type": "4",
              "title": title,
              "caption": caption,
              "usertags": json.dumps({"in": usertags}),
              "date_time_original": date_time_original(time.localtime()),
              "igtv_share_preview_to_feed": "1",
              "upload_id": upload_id,
              "igtv_composer_session_id": self.igtv_composer_session_id,
              "device": self.device,
              "length": duration,
              "clips": [{"length": duration, "source_type": "4"}],
              "extra": {"source_width": width, "source_height": height},
              "audio_muted": False,
              "poster_frame_index": 70,
              **extra_data
          }
          return self.private_request(
              "media/configure_to_igtv/?video=1",
              self.with_default_data(data),
              with_signature=True,
          )
  
  
  def analyze_video(path: Path, thumbnail: Path = None) -> tuple:
      """
      Analyze and crop thumbnail if need
  
      Parameters
      ----------
      path: Path
          Path to the video
      thumbnail: Path
          Path to thumbnail for IGTV
  
      Returns
      -------
      Tuple
          A tuple with (thumbail path, width, height, duration)
      """
  
      print(f'Analizing IGTV file "{path}"')
      width, height, duration = get_data(path)
      if not thumbnail:
          thumbnail = f"{path}.jpg".replace(" ", "")
          print(f'Generating thumbnail "{thumbnail}"...')
          cmd = f'ffmpeg -ss {duration/2} -an -s 404x720 -vframes 1 {thumbnail} -y -i'
          args = shlex.split(cmd)
          args.append(str(path))
          run_cmd(args)
      return thumbnail, width, height, duration
  
  
  def run_cmd(args=[]) -> List:
      return subprocess.check_output(args).decode('utf-8')
  
  def get_sec(duration: str) -> str:
      ts = duration.split('.')[0]
      return sum(int(x) * 60 ** i for i, x in enumerate(reversed(ts.split(':'))))
  
  def get_data(file: Path) -> bool:
      cmd = 'ffprobe -v quiet -print_format json -show_streams'
      args = shlex.split(cmd)
      args.append(file)
      data = json.loads(run_cmd(args))
      height = data['streams'][0]['height']
      width = data['streams'][0]['width']
      duration = data['streams'][-1].get('duration') or get_sec(data['streams'][-1]['tags']['DURATION'])
      return height, width, int(float(duration))
  