在尝试使用Django和Data V3 API在YouTube上发布视频时,如何将用户数据保存到数据库,而不是泡菜或JSON文件

发布于 2025-02-01 08:33:04 字数 4086 浏览 4 评论 0原文

我正在尝试使用Django和MSSQL将视频上传到YouTube,我想将用户数据存储到DB,以便可以从多个帐户登录并发布视频。

YouTube提供的官方文档实现了文件系统,登录后,所有用户数据都将保存在那里,我不想将任何数据存储在文件中,因为将文件保存到DB将是一个巨大的风险,而不是一个好的实践。那么,当我想将视频发布到特定帐户时,如何绕过此步骤并将数据直接保存到DB并检索它?

简而言之,我想用将其存储在数据库中。

这是我的代码

def youtubeAuthenticate():
        os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
        api_service_name = "youtube"
        api_version = "v3"
        client_secrets_file = "client_secrets.json"
        creds = None
        # the file token.pickle stores the user's access and refresh tokens, and is
        # created automatically when the authorization flow completes for the first time
        if os.path.exists("token.pickle"):
            with open("token.pickle", "rb") as token:
                creds = pickle.load(token)
        # if there are no (valid) credentials availablle, let the user log in.
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(client_secrets_file, SCOPES)
                creds = flow.run_local_server(port=0)
            # save the credentials for the next run
            with open("token.pickle", "wb") as token:
                pickle.dump(creds, token)

        return build(api_service_name, api_version, credentials=creds)

@api_view(['GET','POST'])   
def postVideoYT(request):

    youtube = youtubeAuthenticate()
 
    print('yt',youtube)
    try:
        initialize_upload(youtube, request.data)
    except HttpError as e:
        print("An HTTP error %d occurred:\n%s" % (e.resp.status, e.content))

    return Response("Hello")


def initialize_upload(youtube, options):
    

    print('options', options)
    print("title", options['title'])
    # tags = None
    # if options.keywords:
    #     tags = options.keywords.split(",")

    body=dict(
        snippet=dict(
        title=options['title'],
        description=options['description'],
        tags=options['keywords'],
        categoryId=options['categoryId']
        ),
        status=dict(
        privacyStatus=options['privacyStatus']
        )
    )

#   # Call the API's videos.insert method to create and upload the video.

    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,

        media_body=MediaFileUpload(options['file'], chunksize=-1, resumable=True)
    )

    path = pathlib.Path(options['file'])
    ext = path.suffix
    getSize = os.path.getsize(options['file'])
    resumable_upload(insert_request,ext,getSize)

# This method implements an exponential backoff strategy to resume a
# failed upload.
def resumable_upload(insert_request, ext, getSize):
  response = None
  error = None
  retry = 0
  

  while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            if response is not None:
                    respData = response
                    if 'id' in response:
                    print("Video id '%s' was successfully uploaded." % response['id'])
                        
                    else:
                        exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status,
                                                                    e.content)
            else:
                        raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e

        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
                exit("No longer attempting to retry.")

        max_sleep = 2 ** retry
        sleep_seconds = random.random() * max_sleep
        print("Sleeping %f seconds and then retrying..." % sleep_seconds)
        time.sleep(sleep_seconds)
        

I'm trying to upload videos to youtube using Django and MSSQL, I want to store the user data to DB so that I can log in from multiple accounts and post videos.

The official documentation provided by youtube implements a file system and after login, all the user data gets saved there, I don't want to store any data in a file as saving files to DB would be a huge risk and not a good practice. So how can I bypass this step and save data directly to DB and retrieve it when I want to post videos to a specific account?

In short, I want to replace the pickle file implementation with storing it in the database.

Here's my code

def youtubeAuthenticate():
        os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
        api_service_name = "youtube"
        api_version = "v3"
        client_secrets_file = "client_secrets.json"
        creds = None
        # the file token.pickle stores the user's access and refresh tokens, and is
        # created automatically when the authorization flow completes for the first time
        if os.path.exists("token.pickle"):
            with open("token.pickle", "rb") as token:
                creds = pickle.load(token)
        # if there are no (valid) credentials availablle, let the user log in.
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(client_secrets_file, SCOPES)
                creds = flow.run_local_server(port=0)
            # save the credentials for the next run
            with open("token.pickle", "wb") as token:
                pickle.dump(creds, token)

        return build(api_service_name, api_version, credentials=creds)

@api_view(['GET','POST'])   
def postVideoYT(request):

    youtube = youtubeAuthenticate()
 
    print('yt',youtube)
    try:
        initialize_upload(youtube, request.data)
    except HttpError as e:
        print("An HTTP error %d occurred:\n%s" % (e.resp.status, e.content))

    return Response("Hello")


def initialize_upload(youtube, options):
    

    print('options', options)
    print("title", options['title'])
    # tags = None
    # if options.keywords:
    #     tags = options.keywords.split(",")

    body=dict(
        snippet=dict(
        title=options['title'],
        description=options['description'],
        tags=options['keywords'],
        categoryId=options['categoryId']
        ),
        status=dict(
        privacyStatus=options['privacyStatus']
        )
    )

#   # Call the API's videos.insert method to create and upload the video.

    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,

        media_body=MediaFileUpload(options['file'], chunksize=-1, resumable=True)
    )

    path = pathlib.Path(options['file'])
    ext = path.suffix
    getSize = os.path.getsize(options['file'])
    resumable_upload(insert_request,ext,getSize)

# This method implements an exponential backoff strategy to resume a
# failed upload.
def resumable_upload(insert_request, ext, getSize):
  response = None
  error = None
  retry = 0
  

  while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            if response is not None:
                    respData = response
                    if 'id' in response:
                    print("Video id '%s' was successfully uploaded." % response['id'])
                        
                    else:
                        exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status,
                                                                    e.content)
            else:
                        raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e

        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
                exit("No longer attempting to retry.")

        max_sleep = 2 ** retry
        sleep_seconds = random.random() * max_sleep
        print("Sleeping %f seconds and then retrying..." % sleep_seconds)
        time.sleep(sleep_seconds)
        

如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论, 你可以免费 注册 一个本站的账号。
列表为空,暂无数据
我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站,即表示您同意使用 Cookies 和您的相关数据。
原文