cere/python-piped-api-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Compare commits

merge into: cere:master
Branches Tags
cere:master
cere:documentation
cere:v1.0.1-beta
cere:v1.0.0-beta
...
pull from: cere:documentation
Branches Tags
cere:master
cere:documentation
cere:v1.0.1-beta
cere:v1.0.0-beta

No commits in common. "master" and "documentation" have entirely different histories.
master ... documentat

30 changed files with 6089 additions and 1469 deletions
Download patch file Download diff file Expand all files Collapse all files

38
.github/workflows/build_documentation.yml vendored
View file

@ -1,38 +0,0 @@
name: Build documentation
on:
workflow_dispatch:
release:
types: [published]
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10']
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install pdoc
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Build docs
run: |
chmod +x ./scripts/build_documentation.sh
bash ./scripts/build_documentation.sh
- name: Deploy
uses: JamesIves/github-pages-deploy-action@4.1.4
with:
branch: documentation
folder: ./documentation

37
.github/workflows/publish_to_pypi.yml vendored
View file

@ -1,37 +0,0 @@
name: Upload Python Package to PyPI
on:
release:
types: [published]
jobs:
deploy:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10']
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Build package
run: |
chmod +x ./scripts/compile_for_pypi.sh
./scripts/compile_for_pypi.sh
- name: Publish package
uses: pypa/gh-action-pypi-publish@27b31702a0e7fc50959f5ad993c78deac1bdfc29
with:
user: __token__
password: ${{ secrets.PYPI_API_TOKEN }}

28
.github/workflows/pytest.yml vendored
View file

@ -1,28 +0,0 @@
name: Test with pyTest
on: [workflow_dispatch, push, pull_request]
jobs:
deploy:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10']
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Test with pyTest
run: |
python -m pytest -v

9
.gitignore vendored
View file

@ -1,9 +0,0 @@
.vscode/
__pycache__/
env/
.pytest-cache/
documentation/
dist/*
*.egg-info

21
LICENSE
View file

@ -1,21 +0,0 @@
MIT License
Copyright (c) 2021 SKevo
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

47
README.md
View file

@ -1,47 +0,0 @@
# Piped API client (Python)
A Python API wrapper for [Piped](https://piped-docs.kavin.rocks/). This can essentially be used as an alternative way to access YouTube's API, without needing to use an API key. Forked to allow for search queries to be made.
## Installation
```bash
pip install git+https://gitdab.com/cere/python-piped-api-client
```
## Quickstart
Getting started is very easy:
```python
from piped_api import PipedClient
CLIENT = PipedClient()
# Print out the first audio stream URL for a video:
video = CLIENT.get_video(video_id)
audio_stream = video.get_streams('audio')[0]
print(f"Audio stream URL: {audio_stream.url} ({audio_stream.mime_type})")
```
You can find more examples in the [`tests`](https://github.com/CWKevo/python-piped-api-client/tree/master/tests) folder.
## Why?
<!-- Soon... maybe.
This package has allowed me to start creating my open-source project, [ArchiveTube](https://github.com/CWKevo/ArchiveTube) - a scrapper and archive for YouTube content (videos and comments) - to preserve them and make them available to anyone, with ability to search for comments and videos. View hall of fame (most liked comments and videos), bring back dislikes via [ReturnYouTubeDislike.com](https://returnyoutubedislike.com), view deleted content and much more!
Google has showed us that they make YouTube own us by harvesting our data. This is also followed by non-throught out decisions, which their users aren't happy with. Let's do it the other way around this time by reclaiming our content and entertainment back & make YouTube great again!
-->
The creation of this package was primarily fueled by the same type of motivation [Piped has](https://piped-docs.kavin.rocks/docs/why/).
Google's API is not very easy-to-use - you must obtain some JSON thingy to use it, and it is very low-level and not very user-friendly.
On the other hand, this package accessed the [Piped API](https://piped.kavin.rocks/), which has a much more high-level API and doesn't need an account or API keys.
It is not meant to be a replacement for the official YouTube API, but it can help you to cut the strings that Google attaches to you when using their API.
## Useful links
- [Piped's official API documentation](https://piped-docs.kavin.rocks/docs/api-documentation/)

7
index.html Normal file
View file

@ -0,0 +1,7 @@
<!doctype html>
<html>
<head>
<meta charset="utf-8">
<meta http-equiv="refresh" content="0; url=./piped_api.html"/>
</head>
</html>

379
piped_api.html Normal file
View file

File diff suppressed because one or more lines are too long

25
piped_api/__init__.py
View file

@ -1,25 +0,0 @@
import typing as t
from pathlib import Path
from .client import PipedClient
from .models.comments import Comments
# For pdoc:
README_PATH = Path(__file__).parent.parent.absolute() / Path('README.md')
try:
with open(README_PATH, 'r', encoding="UTF-8") as readme:
__readme__ = readme.read()
except:
__readme__ = "Failed to read README.md!"
__doc__ = __readme__
# Supress unused-import warnings:
if t.TYPE_CHECKING:
_ = [PipedClient, Comments]

938
piped_api/client.html Normal file
View file

File diff suppressed because one or more lines are too long

170
piped_api/client.py
View file

@ -1,170 +0,0 @@
import typing as t
from requests import Session
from .models import BasePipedModel
from .models.comments import Comments
from .models.videos import Video
from .models.channels import NextPageChannel, Channel
_MDL = t.TypeVar('_MDL', bound=t.Type[BasePipedModel])
class APIError(Exception): """Raised when an API call fails"""
class PipedClient:
"""
An API client for [Piped](https://piped.kavin.rocks).
See also [Piped API docs](https://piped-docs.kavin.rocks/docs)
"""
def __init__(self, base_api_url: str='https://pipedapi.kavin.rocks', session: t.Type[Session]=Session()) -> None:
"""
### Parameters:
- `base_api_url` - The base URL to the instance's API. Trailing slashes will be stripped.
- `session` - A class/subclass of `requests.Session` to use for all requests.
For example, you could use [requests-cache](https://pypi.org/project/requests-cache/) to make all requests cacheable.
"""
self.base_api_url = base_api_url.strip("/")
self.session = session
def _get_json(self, uri: str, as_model: t.Optional[_MDL]=None, **kwargs) -> t.Union[_MDL, t.Dict[str, t.Any], t.List[t.Any]]:
"""
Obtains JSON data from specific URI of the Piped API.
### Parameters:
- `uri` - The URI to get JSON data from
- `as_model` - The `BasePipedModel` to load the JSON data into. If this is `None`, the JSON data is returned as a `dict`.
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
"""
json: t.Union[dict, list] = self.session.get(f"{self.base_api_url}{uri}", **kwargs).json()
if isinstance(json, dict) and json.get('error', None) is not None:
raise APIError(f"Error: {json['error']}")
if as_model is not None:
return as_model(json)
return json
def get_video(self, video_id: str, **kwargs) -> Video:
"""
Gets information about a specific video.
### Parameters:
- `video_id` - The ID of the video to get information for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#streamsvideoid)
"""
return self._get_json(f"/streams/{video_id}", Video, **kwargs)
def get_comments(self, video_id: str, nextpage: t.Optional[t.Dict[str, t.Optional[str]]]=None, **kwargs) -> Comments:
"""
Gets a list of comments for a specific video.
### Parameters:
- `video_id` - The ID of the video to get comments for
- `nextpage` - Nextpage data, obtained from `.models.comments.Comments.nextpage` property. If this is `None`, the first page of comments is returned.
There are often 20 comments per page.
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#commentsvideoid)
"""
if nextpage is not None:
kwargs.update({'params': {'nextpage': nextpage}})
return self._get_json(f"/nextpage/comments/{video_id}", Comments, **kwargs)
return self._get_json(f"/comments/{video_id}", Comments, **kwargs)
def get_trending(self, country_code: str='US', **kwargs) -> t.List[Video.RelatedStream]:
"""
Obtains trending videos for a specific country. If there are no trending videos (or `country_code` is invalid),
an empty list is returned.
### Parameters:
- `country_code` - The country code ([ISO 3166-1 alpha-2](https://en.wikipedia.org/wiki/ISO_3166-1_alpha-2#Officially_assigned_code_elements)) to get trending videos for. This is automatically capitalized by this package,
since Piped for some reason doesn't accept lowercase country codes. Note: countries such as China or North Korea don't have trending videos, so they will always return an empty list.
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#trending)
"""
kwargs.update({'params': {'region': country_code.upper()}})
return [Video.RelatedStream(trending_video) for trending_video in self._get_json(f"/trending", **kwargs)]
def get_channel_by_id(self, channel_id: str, nextpage: t.Optional[t.Dict[str, t.Optional[str]]]=None, **kwargs) -> t.Union[NextPageChannel, Channel]:
"""
Gets information about a specific channel by its ID.
### Parameters:
- `channel_id` - The ID of the channel to get information for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#channelchannelid)
"""
if nextpage is not None:
kwargs.update({'params': {'nextpage': nextpage}})
return self._get_json(f"/nextpage/channel/{channel_id}", NextPageChannel, **kwargs)
return self._get_json(f"/channel/{channel_id}", Channel, **kwargs)
def get_channel_by_name(self, channel_name: str, **kwargs) -> Channel:
"""
Gets information about a specific channel by its name.
### Parameters:
- `channel_name` - The name of the channel to get information for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#cname)
"""
return self._get_json(f"/c/{channel_name}", Channel, **kwargs)
def get_search_suggestions(self, search_query: str, **kwargs) -> t.List[str]:
"""
Obtains search suggestions for a query.
### Parameters:
- `search_query` - The query to get search suggestions for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
[Piped Documentation](https://piped-docs.kavin.rocks/docs/api-documentation/#suggestions)
"""
kwargs.update({'params': {'query': search_query}})
return self._get_json(f"/suggestions", **kwargs)
def get_search_results(self, search_query: str, filter: str="all", **kwargs) -> t.List[str]:
"""
Obtains search results for a query.
### Parameters:
- `search_query` - The query to get search results for
- `**kwargs` - Additional keyword arguments to pass to `requests.Session.get`
"""
kwargs.update({'params': {'q': search_query, 'filter': filter}})
return self._get_json(f"/search", **kwargs)

327
piped_api/models.html Normal file
View file

File diff suppressed because one or more lines are too long

15
piped_api/models/__init__.py
View file

@ -1,15 +0,0 @@
import typing as t
class BasePipedModel:
"""
Base class for all Piped models.
"""
def __init__(self, data: t.Dict[str, t.Any]) -> None:
"""
### Parameters:
- `data` - The JSON (`dict`) data to initialize the model with.
"""
self.data = data

651
piped_api/models/channels.html Normal file
View file

File diff suppressed because one or more lines are too long

104
piped_api/models/channels.py
View file

@ -1,104 +0,0 @@
import typing as t
from . import BasePipedModel
from .videos import Video
class NextPageChannel(BasePipedModel):
"""
Represents a channel obtained via the `nextpage` endpoint.
This model contains only `nextpage` and `relatedStreams`. It's a parent for `Channel`.
"""
@property
def nextpage(self) -> str:
"""
A JSON encoded string to be passed to the `'nextpage'` endpoint(s) when
obtaining paginated data.
"""
return self.data['nextpage']
@property
def uploaded_videos(self) -> t.List[Video.RelatedStream]:
"""
List of uploaded videos from the current fetched data
There are max. 30 videos per page
"""
return [Video.RelatedStream(video_data) for video_data in self.data['relatedStreams']]
class Channel(NextPageChannel):
"""
Represents a YouTube channel.
Contains properties of `NextPageChannel`.
"""
@property
def id(self) -> str:
"""
The channel's ID
"""
return self.data['id']
@property
def name(self) -> str:
"""
The channel's name
"""
return self.data['name']
@property
def avatar_url(self) -> str:
"""
The channel's avatar URL
"""
return self.data['avatarUrl']
@property
def banner_url(self) -> str:
"""
The channel's banner URL
"""
return self.data['bannerUrl']
@property
def description(self) -> str:
"""
The channel's description
"""
return self.data['description']
@property
def subscriber_count(self) -> int:
"""
The number of subscribers the channel has
"""
return self.data['subscriberCount']
@property
def verified(self) -> bool:
"""
Whether or not the channel is verified by YouTube (has a badge)
"""
return self.data['verified']

1002
piped_api/models/comments.html Normal file
View file

File diff suppressed because one or more lines are too long

163
piped_api/models/comments.py
View file

@ -1,163 +0,0 @@
import typing as t
from . import BasePipedModel
class Comments(BasePipedModel):
class Comment(BasePipedModel):
@property
def author(self) -> str:
"""
The name of the author of the comment
"""
return self.data['author']
@property
def comment_id(self) -> str:
"""
The comment ID
"""
return self.data['commentId']
@property
def comment_text(self) -> str:
"""
The text of the comment
"""
return self.data['commentText']
@property
def commented_time(self) -> str:
"""
The time the comment was made (format: `'x y ago'`).
### Note:
The raw time from API also includes the `'(edited)'` suffix to mark comment as edited (if it was).
By accessing this property, the suffix is automatically removed.
If you for whatever reason want to keep the suffix, access this property directly via `Comment.data['commentedTime']`
"""
time: str = self.data['commentedTime']
return time.removesuffix(' (edited)')
@property
def is_edited(self) -> bool:
"""
Whether or not the comment is edited.
### Note:
This property checks whether there is `'(edited)'` in the `commentedTime` property, because that's where you get that from.
See `Comments.Comment.commented_time`
"""
time: str = self.data['commentedTime']
return time.endswith('(edited)')
@property
def commentor_url(self) -> str:
"""
The URL of the channel that made the comment
"""
return self.data['commentorUrl']
@property
def replies_page(self) -> t.Optional[str]:
"""
Same as `Comments.nextpage`, but to load replies.
`None` means that there are no replies.
"""
return self.data['repliesPage']
@property
def hearted(self) -> bool:
"""
Whether or not the comment has been hearted
"""
return self.data['hearted']
@property
def like_count(self) -> int:
"""
The number of likes the comment has
"""
return self.data['likeCount']
@property
def pinned(self) -> bool:
"""
Whether or not the comment is pinned
"""
return self.data['pinned']
@property
def thumbnail(self) -> str:
"""
The thumbnail of the commentor's channel
"""
return self.data['thumbnail']
@property
def verified(self) -> bool:
"""
Whether or not the author of the comment is verified
"""
return self.data['verified']
def get_comments(self) -> t.List[Comment]:
"""
Obtain a list of comments
"""
return [self.Comment(comment_json) for comment_json in self.data['comments']]
def __iter__(self) -> t.Iterator[Comment]:
iter(self.get_comments())
@property
def disabled(self) -> bool:
"""
Whether or not the comments are disabled
"""
return self.data['disabled']
@property
def nextpage(self) -> t.Optional[str]:
"""
A JSON encoded page, which is used for the nextpage endpoint.
If there is no nextpage data, this returns `None`.
"""
return self.data['nextpage']

2739
piped_api/models/videos.html Normal file
View file

File diff suppressed because one or more lines are too long

559
piped_api/models/videos.py
View file

@ -1,559 +0,0 @@
import typing as t
from datetime import datetime, date, timedelta
from . import BasePipedModel
class Video(BasePipedModel):
@property
def title(self) -> str:
"""
The title/name of the video
"""
return self.data['title']
@property
def description(self) -> str:
"""
The description of the video
"""
return self.data['description']
@property
def upload_date(self) -> date:
"""
The date the video was uploaded at.
### Note:
Use `Video.data['uploadDate']` to get the raw string that was obtained from the API - this package
automatically converts the raw string to a `datetime.date` object.
"""
raw = self.data['uploadDate']
return datetime.strptime(raw, r"%Y-%m-%d").date()
@property
def uploader(self) -> str:
"""
The channel name of the author of the video
"""
return self.data['uploader']
@property
def uploader_url(self) -> str:
"""
The URI to the author's channel
"""
return self.data['uploaderUrl']
@property
def uploader_avatar(self) -> str:
"""
The URL to the video author's avatar image
"""
return self.data['uploaderAvatar']
@property
def thumbnail_url(self) -> str:
"""
The URL to the video's thumbnail image
"""
return self.data['thumbnail']
@property
def hls(self) -> t.Optional[str]:
"""
The hls manifest URL, to be used for Livestreams
"""
return self.data['hls']
@property
def dash(self) -> t.Optional[str]:
"""
The dash manifest URL for OTF streams
"""
return self.data['dash']
@property
def lbry_id(self) -> str:
"""
The lbry id of the video, if available. I assume this has something to do with https://lbry.com/
"""
return self.data['lbryId']
@property
def uploader_verified(self) -> str:
"""
Whether or not the channel that uploaded the video is verified by YouTube (badge)
"""
return self.data['uploaderVerified']
@property
def duration(self) -> timedelta:
"""
The duration of the video.
### Note:
The original value from the API was in seconds (`Video.data['duration']`), but this package
converts it to a `datetime.timedelta` object.
"""
return timedelta(seconds=self.data['duration'])
@property
def views(self) -> int:
"""
The number of views the video has received
"""
return self.data['views']
@property
def likes(self) -> int:
"""
The amount of likes the video has received. `-1` if hidden
"""
return self.data['likes']
@property
def dislikes(self) -> int:
"""
The amount of dislikes the video has received. `-1` if hidden
### Note:
This is obsolete since YouTube did a tiny gigantical little big whoopsie with their like system and screwed it all up
You can use awesome user-made projects such as https://returnyoutubedislike.com to obtain the dislike count
"""
return self.data['dislikes']
class Stream(BasePipedModel):
"""
Either an audio or video stream of a video
"""
@property
def url(self) -> str:
"""
The URL of the stream
"""
return self.data['url']
@property
def format(self) -> str:
"""
The format of the stream (`'M4A' or 'WEBMA_OPUS' or 'MPEG_4' or 'WEBM' or 'v3GPP'`
No, I don't know how many are there or what does each mean
"""
return self.data['format']
@property
def quality(self) -> str:
"""
The standard quality we all know and love (e. g.: `'240p'` for video or `'128k'` for audio)
"""
return self.data['quality']
@property
def mime_type(self) -> str:
"""
If you come from web development (or other invidious area that works with these French mimes),
then you already know what this is. If not, consider [checking the Mozilla documentation](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
"""
return self.data['mimeType']
@property
def codec(self) -> str:
"""
What is this? I don't know. A codec?
"""
return self.data['codec']
@property
def video_only(self) -> bool:
"""
Whether or not the stream is video only (AKA. muted video)
"""
return self.data['videoOnly']
@property
def bitrate(self) -> int:
"""
The bitrate of the stream
"""
return self.data['bitrate']
@property
def init_start(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['initStart']
@property
def init_end(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['initEnd']
@property
def index_start(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['indexStart']
@property
def index_end(self) -> int:
"""
Not sure what this does, but it seems to be useful for creating dash streams
"""
return self.data['indexEnd']
@property
def width(self) -> int:
"""
The width of the stream. `'0'` for audio streams (makes sense)
"""
return self.data['width']
@property
def height(self) -> int:
"""
The height of the stream. `'0'` for audio streams (makes sense)
"""
return self.data['width']
@property
def fps(self) -> int:
"""
Frames Per Second. This is usually `'0'` for audio and `'30'` or `'60'` for video
"""
return self.data['fps']
def get_streams(self, type: t.Literal['video', 'audio']='video') -> t.List[Stream]:
"""
Get the streams of a video.
### Parameters:
- `type` - The type of stream to get. Either `'video'` or `'audio'`
"""
if type == 'video' or type == 'audio':
return [self.Stream(stream_data) for stream_data in self.data[f"{type}Streams"]]
raise ValueError('Invalid stream type. Must be either `video` or `audio`')
class RelatedStream(BasePipedModel):
"""
A related stream (e. g.: related video to the current one from the right sidebar, video related to/uploaded by a channel and trending video).
"""
@property
def url(self) -> str:
"""
The URL to the related video
"""
return self.data['url']
@property
def title(self) -> str:
"""
The title of the related video
"""
return self.data['title']
@property
def thumbnail(self) -> str:
"""
The thumbnail URL of the related video
"""
return self.data['thumbnail']
@property
def uploader_name(self) -> str:
"""
The name of the channel that uploaded the related video
"""
return self.data['uploaderName']
@property
def uploader_url(self) -> str:
"""
The URL of the channel that uploaded the related video
"""
return self.data['uploaderUrl']
@property
def uploader_avatar(self) -> str:
"""
The URL of the channel's avatar
"""
return self.data['uploaderAvatar']
@property
def uploaded_date(self) -> str:
"""
The date the related video was uploaded (format: `'x y ago'`)
"""
return self.data['uploadedDate']
@property
def short_description(self) -> t.Optional[str]:
"""
The short description of the related video. As far as I know, this is always `None` - perhaps some unused YouTube feature?
"""
return self.data['shortDescription']
@property
def duration(self) -> timedelta:
"""
The duration of the related video.
### Note:
The original value from the API was in seconds (`Video.data['duration']`), but this package
converts it to a `datetime.timedelta` object.
"""
return timedelta(seconds=self.data['duration'])
@property
def views(self) -> str:
"""
The amount of views the related video has received
"""
return self.data['views']
@property
def uploaded(self) -> datetime:
"""
The date the related video was uploaded (as a `datetime.datetime` object).
### Note:
The original value was in milliseconds since epoch (`Video.data['uploaded']`), but this package converts it to a `datetime.datetime` object.
"""
return datetime.fromtimestamp(self.data['uploaded'] / 1000)
@property
def uploader_verified(self) -> bool:
"""
Whether or not the channel that uploaded the related video is verified by YouTube (e. g.: has badge)
"""
return self.data['uploaderVerified']
@property
def related_videos(self) -> t.List[RelatedStream]:
"""
List of related streams
"""
return [self.RelatedStream(video_data) for video_data in self.data['relatedStreams']]
class Subtitle(BasePipedModel):
@property
def url(self) -> str:
"""
The URL to the subtitle
"""
return self.data['url']
@property
def mime_type(self) -> str:
"""
If you come from web development (or other invidious area that works with these French mimes),
then you already know what this is. If not, consider [checking the Mozilla documentation](https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types)
"""
return self.data['mimeType']
@property
def name(self) -> str:
"""
The name of the language the captions are in
"""
return self.data['name']
@property
def code(self) -> str:
"""
The country code for the captions
"""
return self.data['code']
@property
def auto_generated(self) -> bool:
"""
Whether or not the captions are auto-generated by YouTube
"""
return self.data['autoGenerated']
@property
def subtitles(self) -> t.List[Subtitle]:
"""
A list of captions for the video
"""
return [self.Subtitle(subtitle_data) for subtitle_data in self.data['subtitles']]
@property
def livestream(self) -> bool:
"""
Whether or not the video is a livestream
"""
return self.data['livestream']
@property
def proxy_url(self) -> str:
"""
The base URL for Piped proxy
"""
return self.data['proxyUrl']
class Chapter(BasePipedModel):
"""
A video chapter (or "section").
YouTube displays a list of chapters, if there are timestamps in the description.
"""
@property
def title(self) -> str:
"""
The title of the chapter
"""
return self.data['title']
@property
def image(self) -> str:
"""
The image URL for the chapter
"""
return self.data['image']
@property
def start(self) -> timedelta:
"""
The start time of the chapter
### Note:
The original value from the API was in seconds, this package automatically converts it to `datetime.timedelta`
"""
return timedelta(seconds=self.data['start'])
@property
def chapters(self) -> t.List[Chapter]:
"""
A list of chapters for the video
"""
return [self.Chapter(chapter_data) for chapter_data in self.data['chapters']]

3
requirements.txt
View file

@ -1,3 +0,0 @@
pytest
requests

4
scripts/build_documentation.sh
View file

@ -1,4 +0,0 @@
cd "$(dirname "$0")"
cd ../
pdoc -o documentation piped_api

4
scripts/compile_everything.sh
View file

@ -1,4 +0,0 @@
cd "$(dirname "$0")"
bash ./build_documentation.sh
bash ./compile_for_pypi.sh

4
scripts/compile_for_pypi.sh
View file

@ -1,4 +0,0 @@
cd "$(dirname "$0")"
cd ../
python setup.py sdist

46
search.js Normal file
View file

File diff suppressed because one or more lines are too long

62
setup.py
View file

@ -1,62 +0,0 @@
from pathlib import Path
import sys
sys.path.append('.')
import setuptools
__description__ = "Piped API client"
__author__ = "SKevo"
__copyright__ = "Copyright (c) 2021, SKevo"
__credits__ = ["SKevo"]
__license__ = "MIT"
__version__ = "v1.0.1-beta"
__maintainer__ = "SKevo"
__email__ = "me@kevo.link"
__status__ = "4 - Beta"
README_PATH = Path(__file__).parent.absolute() / Path('README.md')
try:
with open(README_PATH, 'r', encoding="UTF-8") as readme:
__readme__ = readme.read()
except:
__readme__ = "Failed to read README.md!"
__doc__ = __readme__
setuptools.setup(
name = 'piped-api',
packages = setuptools.find_packages(exclude=('tests',)),
long_description=__readme__,
long_description_content_type='text/markdown',
version = __version__,
license = __license__,
description = __description__,
keywords = ["piped", "api", "client"],
author = __author__,
author_email = __email__,
url = 'https://github.com/CWKevo/python-piped-api-client',
install_requires=['requests'],
classifiers=[
f'Development Status :: {__status__}',
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
],
)

4
tests/__init__.py
View file

@ -1,4 +0,0 @@
from piped_api import PipedClient
CLIENT = PipedClient()

58
tests/test_channels.py
View file

@ -1,58 +0,0 @@
from tests import CLIENT
from datetime import timedelta
from random import choice
def test_channel_by_id(channel_id: str='UCuAXFkgsw1L7xaCfnd5JJOw') -> None:
"""
Prints out information about a channel by its ID.
"""
channel = CLIENT.get_channel_by_id(channel_id)
video_id = channel.uploaded_videos[0].url.removeprefix('/watch?v=')
likes = CLIENT.get_video(video_id).likes
print(f"Total likes for last video of {channel.name}: {likes}")
def test_channel_by_name(channel_name: str='SusanWojcicki') -> None:
"""
Prints out information about a channel by its ID.
"""
channel = CLIENT.get_channel_by_name(channel_name)
print(f"""
Channel ID: {channel.id}
Name: {channel.name}
Description: {channel.description}
Subscriber count: {channel.subscriber_count}
""")
def test_get_watchtime_trending() -> None:
"""
Prints out the total watchtime for recent videos of a random trending channel
"""
trending_channel_id = choice(CLIENT.get_trending('SK')).uploader_url.removeprefix('/channel/')
trending_channel = CLIENT.get_channel_by_id(trending_channel_id)
total_watchtime = timedelta(milliseconds=0)
for video in trending_channel.uploaded_videos:
total_watchtime += video.duration
print(f"Total watchtime for recent {len(trending_channel.uploaded_videos)} videos of {trending_channel.name} (https://youtube.com/channel/{trending_channel.id}): {total_watchtime}")
if __name__ == '__main__':
test_channel_by_name()
test_channel_by_id()
test_get_watchtime_trending()

34
tests/test_comments.py
View file

@ -1,34 +0,0 @@
from tests import CLIENT
def test_comments(video_id: str='dQw4w9WgXcQ') -> None:
"""
Prints out first 20 pages of comments from a video.
"""
at_page = 0
max_pages = 5
total_comments = 0
np = None
while at_page < max_pages:
comments = CLIENT.get_comments(video_id, nextpage=np, params={'hl': 'us'})
at_page += 1
print('=' * 35, f'Page: {at_page}', '=' * 35)
for comment in comments.get_comments():
total_comments += 1
print(f'Comment {comment.comment_id} by "{comment.author}" ({comment.commented_time}), {comment.like_count} likes: "{comment.comment_text}"')
if comments.nextpage == None:
print(f"No more comments! Total: {total_comments}, expected: {max_pages * 20}")
break
np = comments.nextpage
print(f"Okay, that's enough comments... Total: {total_comments}, expected: {max_pages * 20}")
if __name__ == '__main__':
test_comments()

17
tests/test_suggestions.py
View file

@ -1,17 +0,0 @@
from tests import CLIENT
def test_suggestions(search_query: str='Susan') -> None:
"""
Obtains search suggestions for a query.
"""
suggestions = CLIENT.get_search_suggestions(search_query)
assert len(suggestions) > 0
print(suggestions)
if __name__ == '__main__':
test_suggestions()

63
tests/test_videos.py
View file

@ -1,63 +0,0 @@
import typing as t
from tests import CLIENT
from datetime import datetime
def test_video(video_id: str='dQw4w9WgXcQ') -> None:
"""
Prints out information about a video.
"""
video = CLIENT.get_video(video_id)
short_description = video.description[:100].replace('\n', '')
print(f"""
Video ID: {video_id}
Title: {video.title}
Description: {short_description}...
Views: {video.views}
Uploaded by: {video.uploader}
Uploaded on: {video.upload_date} ({datetime.now().year - video.upload_date.year} years ago)
Duration: {video.duration}
FPS: {video.get_streams('video')[0].fps}
""")
def test_trending(country_codes: t.List[str]=['US', 'SK', 'CN']) -> None:
"""
Prints out trending videos for a specific country.
"""
for country_code in country_codes:
videos = CLIENT.get_trending(country_code)
# Nothing ever trends in China's YouTube:
if country_code == 'CN':
assert len(videos) == 0
print("\nYes, empty list works.")
for video in videos:
print(f"{video.uploader_name} >> {video.title} ({video.views} views)")
def test_get_audio(video_id: str='dQw4w9WgXcQ') -> None:
"""
Prints out the first audio stream URL for a video.
"""
video = CLIENT.get_video(video_id)
audio_stream = video.get_streams('audio')[0]
print(f"Audio stream URL: {audio_stream.url} ({audio_stream.mime_type})")
if __name__ == '__main__':
test_video()
test_trending()
test_get_audio()