Release of old sources

This commit is contained in:
Andrew 2026-02-08 16:05:45 +07:00
commit f3e1fa4222
6 changed files with 214 additions and 0 deletions

3
README.md Normal file
View file

@ -0,0 +1,3 @@
# HIVOD
My old project made for automation of uploading twitch VODs to youtube

45
api.py Normal file
View file

@ -0,0 +1,45 @@
import requests
import httplib2
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from oauth2client.client import AccessTokenCredentials
from settings import *
from downloader import resumable_upload
def get_auth_code():
""" Get access token for connect to youtube api """
oauth_url = 'https://accounts.google.com/o/oauth2/token'
data = {
"refresh_token": YOUTUBE_REFRESH_TOKEN,
"client_id": YOUTUBE_CLIENT_ID,
"client_secret": YOUTUBE_CLIENT_SECRET,
"grant_type": "refresh_token",
}
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json'
}
response = requests.post(oauth_url, data=data, headers=headers)
return response.json()['access_token']
def get_authenticated_service():
""" Create youtube oauth2 connection """
credentials = AccessTokenCredentials(
access_token=get_auth_code(), user_agent='my-awesome-project/1.0'
)
return build('youtube', 'v3', http=credentials.authorize(httplib2.Http()))
def initialize_upload(youtube, metadata: dict, video_path: str):
""" Create youtube upload data """
# create video meta data
# Call the API's videos.insert method to create and upload the video
insert_request = youtube.videos().insert(
part=",".join(metadata.keys()), body=metadata,
media_body=MediaFileUpload(video_path, chunksize=-1, resumable=True))
# wait for file uploading
return resumable_upload(insert_request)

49
downloader.py Normal file
View file

@ -0,0 +1,49 @@
import random
import http
import time
import httplib2
from googleapiclient.errors import HttpError
# Explicitly tell the underlying HTTP transport library not to retry, since we are handling retry logic ourselves.
httplib2.RETRIES = 1
# Maximum number of times to retry before giving up.
MAX_RETRIES = 10
# Always retry when these exceptions are raised.
RETRIABLE_EXCEPTIONS = (
http.HTTPStatus.INSUFFICIENT_STORAGE,
httplib2.HttpLib2Error, IOError, http.client.NotConnected,
http.client.IncompleteRead, http.client.ImproperConnectionState,
http.client.CannotSendRequest, http.client.CannotSendHeader,
http.client.ResponseNotReady, http.client.BadStatusLine)
# Always retry when an HttpError with one of these status codes is raised.
RETRIABLE_STATUS_CODES = (500, 502, 503, 504)
def resumable_upload(insert_request):
response = None
error = None
retry = 0
while response is None:
try:
status, response = insert_request.next_chunk()
if 'id' in response:
return response['id']
except HttpError as err:
if err.resp.status in RETRIABLE_STATUS_CODES:
error = True
else:
raise
except RETRIABLE_EXCEPTIONS:
error = True
if error:
retry += 1
if retry > MAX_RETRIES:
raise Exception('Maximum retry are fail')
sleep_seconds = random.random() * 2 ** retry
time.sleep(sleep_seconds)

67
hivod.py Normal file
View file

@ -0,0 +1,67 @@
import os
import argparse
import youtube_dl
from api import get_authenticated_service, initialize_upload
from settings import Categories, Privacy
SIMULATE = False
def generate_metadata(title: str, description: str, tags: list, category_id: int, privacy_status: str) -> dict:
return {
"snippet": {
"title": title,
"description": description,
"tags": tags,
"categoryId": category_id
},
"status": {
"privacyStatus": privacy_status
}
}
def main(args):
ydl_opts = {"simulate": SIMULATE}
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
result = ydl.extract_info(args.vod, download=True)
title = "[VOD] {}".format(result["title"])
u_date = result["upload_date"]
description = "Date: {}.{}.{}\n".format(u_date[6:], u_date[4:6], u_date[:4]) + \
"Streamer: {}\n".format(result["uploader"]) + \
"Support streamer: https://www.twitch.tv/{}\n\n".format(result["uploader_id"]) + \
"VIDEO IS MONETISED NOT BY ME"
fname = "./" + ydl.prepare_filename(result)
if args.title:
title = args.title
if args.description:
description = args.description
print("=====File will be uploaded with this parameters=====")
print("Filename:", fname)
print("Title:", title)
print("Description:", description)
print("====================================================")
metadata = generate_metadata(title, description, [], Categories.ENTERTAINMENT, Privacy.PUBLIC)
video_path = os.path.abspath(fname)
try:
video_id = initialize_upload(get_authenticated_service(), metadata, video_path)
except Exception as e:
print(e)
else:
print("Uploaded video id:", video_id)
print("Removing video file...", end="")
os.unlink(video_path)
print("done")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="High velocity VOD uploader")
parser.add_argument("vod", type=str, help="link to VOD")
parser.add_argument("--title", type=str, help="custom video title")
parser.add_argument("--description", type=str, help="custom video description")
main(parser.parse_args())

21
requirements.txt Normal file
View file

@ -0,0 +1,21 @@
cachetools==4.1.0
certifi==2020.4.5.2
chardet==3.0.4
google-api-core==1.20.0
google-api-python-client==1.9.2
google-auth==1.16.1
google-auth-httplib2==0.0.3
googleapis-common-protos==1.52.0
httplib2==0.18.1
idna==2.9
oauth2client==4.1.3
protobuf==3.12.2
pyasn1==0.4.8
pyasn1-modules==0.2.8
pytz==2020.1
requests==2.23.0
rsa==4.0
six==1.15.0
uritemplate==3.0.1
urllib3==1.25.9
youtube-dl==2020.6.6

29
settings.py Normal file
View file

@ -0,0 +1,29 @@
import enum
YOUTUBE_REFRESH_TOKEN = ""
YOUTUBE_CLIENT_ID = ""
YOUTUBE_CLIENT_SECRET = ""
class Categories(object):
FILM_AND_ANIMATION = 1
AUTOS_AND_VEHICLES = 2
MUSIC = 10
PETS_AND_ANIMALS = 15
SPORTS = 17
TRAVEL_AND_EVENTS = 19
GAMING = 20
PEOPLE_AND_BLOGS = 22
COMEDY = 23
ENTERTAINMENT = 24
NEWS_AND_POLITICS = 25
HOWTO_AND_STYLE = 26
EDUCATION = 27
SCIENCE_AND_TECHNOLOGY = 28
class Privacy(object):
PRIVATE = "private"
PUBLIC = "public"
UNLISTED = "unlisted"