from __future__ import annotations
from ..helpers import extract_video_id_from_url
from typing import TYPE_CHECKING, ClassVar, Iterator, Optional
from datetime import datetime
import requests
from ..exceptions import InvalidResponseException
import json
if TYPE_CHECKING:
from ..tiktok import TikTokApi
from .user import User
from .sound import Sound
from .hashtag import Hashtag
from .comment import Comment
[docs]
class Video:
"""
A TikTok Video class
Example Usage
```py
video = api.video(id='7041997751718137094')
```
"""
parent: ClassVar[TikTokApi]
id: Optional[str]
"""TikTok's ID of the Video"""
url: Optional[str]
"""The URL of the Video"""
create_time: Optional[datetime]
"""The creation time of the Video"""
stats: Optional[dict]
"""TikTok's stats of the Video"""
author: Optional[User]
"""The User who created the Video"""
sound: Optional[Sound]
"""The Sound that is associated with the Video"""
hashtags: Optional[list[Hashtag]]
"""A List of Hashtags on the Video"""
as_dict: dict
"""The raw data associated with this Video."""
def __init__(
self,
id: Optional[str] = None,
url: Optional[str] = None,
data: Optional[dict] = None,
**kwargs,
):
"""
You must provide the id or a valid url, else this will fail.
"""
self.id = id
self.url = url
if data is not None:
self.as_dict = data
self.__extract_from_data()
elif url is not None:
i, session = self.parent._get_session(**kwargs)
self.id = extract_video_id_from_url(
url,
headers=session.headers,
proxy=kwargs.get("proxy")
if kwargs.get("proxy") is not None
else session.proxy,
)
if getattr(self, "id", None) is None:
raise TypeError("You must provide id or url parameter.")
[docs]
async def info(self, **kwargs) -> dict:
"""
Returns a dictionary of all data associated with a TikTok Video.
Note: This is slow since it requires an HTTP request, avoid using this if possible.
Returns:
dict: A dictionary of all data associated with a TikTok Video.
Raises:
InvalidResponseException: If TikTok returns an invalid response, or one we don't understand.
Example Usage:
.. code-block:: python
url = "https://www.tiktok.com/@davidteathercodes/video/7106686413101468970"
video_info = await api.video(url=url).info()
"""
i, session = self.parent._get_session(**kwargs)
proxy = (
kwargs.get("proxy") if kwargs.get("proxy") is not None else session.proxy
)
if self.url is None:
raise TypeError("To call video.info() you need to set the video's url.")
r = requests.get(self.url, headers=session.headers, proxies=proxy)
if r.status_code != 200:
raise InvalidResponseException(
r.text, "TikTok returned an invalid response.", error_code=r.status_code
)
# Try SIGI_STATE first
# extract tag <script id="SIGI_STATE" type="application/json">{..}</script>
# extract json in the middle
start = r.text.find('<script id="SIGI_STATE" type="application/json">')
if start != -1:
start += len('<script id="SIGI_STATE" type="application/json">')
end = r.text.find("</script>", start)
if end == -1:
raise InvalidResponseException(
r.text, "TikTok returned an invalid response.", error_code=r.status_code
)
data = json.loads(r.text[start:end])
video_info = data["ItemModule"][self.id]
else:
# Try __UNIVERSAL_DATA_FOR_REHYDRATION__ next
# extract tag <script id="__UNIVERSAL_DATA_FOR_REHYDRATION__" type="application/json">{..}</script>
# extract json in the middle
start = r.text.find('<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__" type="application/json">')
if start == -1:
raise InvalidResponseException(
r.text, "TikTok returned an invalid response.", error_code=r.status_code
)
start += len('<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__" type="application/json">')
end = r.text.find("</script>", start)
if end == -1:
raise InvalidResponseException(
r.text, "TikTok returned an invalid response.", error_code=r.status_code
)
data = json.loads(r.text[start:end])
default_scope = data.get("__DEFAULT_SCOPE__", {})
video_detail = default_scope.get("webapp.video-detail", {})
if video_detail.get("statusCode", 0) != 0: # assume 0 if not present
raise InvalidResponseException(
r.text, "TikTok returned an invalid response structure.", error_code=r.status_code
)
video_info = video_detail.get("itemInfo", {}).get("itemStruct")
if video_info is None:
raise InvalidResponseException(
r.text, "TikTok returned an invalid response structure.", error_code=r.status_code
)
self.as_dict = video_info
self.__extract_from_data()
return video_info
[docs]
async def bytes(self, **kwargs) -> bytes:
"""
Returns the bytes of a TikTok Video.
TODO:
Not implemented yet.
Example Usage:
.. code-block:: python
video_bytes = api.video(id='7041997751718137094').bytes()
# Saving The Video
with open('saved_video.mp4', 'wb') as output:
output.write(video_bytes)
"""
raise NotImplementedError
i, session = self.parent._get_session(**kwargs)
downloadAddr = self.as_dict["video"]["downloadAddr"]
cookies = await self.parent.get_session_cookies(session)
cookie_str = "; ".join([f"{k}={v}" for k, v in cookies.items()])
h = session.headers
h["cookie"] = cookie_str
# Fetching the video bytes using a browser fetch within the page context
file_bytes = await session.page.evaluate(
"""
async (url, headers) => {
const response = await fetch(url, { headers });
if (response.ok) {
const buffer = await response.arrayBuffer();
return new Uint8Array(buffer);
} else {
return `Error: ${response.statusText}`; // Return an error message if the fetch fails
}
}
""",
(downloadAddr, h),
)
byte_values = [
value
for key, value in sorted(file_bytes.items(), key=lambda item: int(item[0]))
]
return bytes(byte_values)
def __extract_from_data(self) -> None:
data = self.as_dict
self.id = data["id"]
timestamp = data.get("createTime", None)
if timestamp is not None:
try:
timestamp = int(timestamp)
except ValueError:
pass
self.create_time = datetime.fromtimestamp(timestamp)
self.stats = data["stats"]
author = data.get("author")
if isinstance(author, str):
self.author = self.parent.user(username=author)
else:
self.author = self.parent.user(data=author)
self.sound = self.parent.sound(data=data)
self.hashtags = [
self.parent.hashtag(data=hashtag) for hashtag in data.get("challenges", [])
]
if getattr(self, "id", None) is None:
Video.parent.logger.error(
f"Failed to create Video with data: {data}\nwhich has keys {data.keys()}"
)
def __repr__(self):
return self.__str__()
def __str__(self):
return f"TikTokApi.video(id='{getattr(self, 'id', None)}')"