Python分享消息到Twitter、Youtube、Facebook平台

xiaoxiao2021-02-28  12

Python分享消息到Twitter、Youtube、Facebook平台

浏览器

概述

第三方登录平台免不了需要登录验证,比起命令行输入账号密码再填充表单,不如直接用浏览器实在

环境配置

运行库
pip install PyQt5 pip install PyQt5-tools
代码
import sys from PyQt5 import QtCore, QtGui, QtWidgets from PyQt5.QtCore import * from PyQt5.QtGui import * from PyQt5.QtWidgets import * from PyQt5.QtWebKit import * from PyQt5.QtWebKitWidgets import * from PyQt5.QtWidgets import QApplication, QWidget, QMainWindow class webForm(): def __init__(self): self.app = QApplication(sys.argv) self.form = QWidget() self._debug = False self.webview = QWebView() self.webview.page().mainFrame().urlChanged.connect(self._urlChanged) vbox = QVBoxLayout() vbox.addWidget(self.webview) main = QGridLayout() main.setSpacing(0) main.addLayout(vbox, 0, 0) self.form.setLayout(main) self.callbackFunc = None def show(self, url, title, debug=False): """ 打开内嵌浏览器(在浏览器窗口关闭前,主线程将被挂起) url: 目标链接 title: 窗口标题 debug: 是否在控制台打印出每次url跳转及参数 """ self._debug = debug self.form.setWindowTitle(title) self.form.show() self.webview.load(QUrl(url)) self.app.exec_() def close(self): self.form.close() def __del__(self): self.close() def set_callback(self, callbackFunc): """ callbackFunc(QUrl) """ self.callbackFunc = callbackFunc def _loadFinished(self,bool): if bool != True: return url = self.webview.url() if self.callbackFunc != None: self.callbackFunc(url) def _urlChanged(self, url): ##debug############# if self._debug == True: query = QUrlQuery(url.query()) items = query.queryItems() logging.info("--------------------------------------") logging.info("[url]: " + url.toString()) logging.info("[host]: " + url.host()) #print(" query: " + query.toString()) for i in items: logging.info(" " + i[0] + " : " + i[1]) ##################### if self.callbackFunc != None: self.callbackFunc(url) pass

Twitter

概述

使用tweepy进行登录验证拿到access_token后,可以自由发布文字、图片。

需要登录验证需要第三方库:tweepy需要建立开发者应用可以发布纯文字推、带图推及视频上传功能在当前版本tweepy(3.6.0)有bug,在github上已经被修复,但还没有发布到新版本

相关链接

twitter 开发者后台 https://dev.twitter.com/appstwitter 授权过程 https://blog.csdn.net/yangjian8915/article/details/11816669twitter API索引 https://developer.twitter.com/en/docs/api-reference-indextweepy github https://github.com/tweepy/tweepy

环境配置

开发者后台
打开https://apps.twitter.com/ 新建应用依次输入相关信息。其中:Callback URLs必须是可访问的地址进入应用,点开Keys and Access Tokens得到Consumer Key和Consumer Secret
运行库

pip install tweepy

代码
import tweepy import logging class Twitter(): def __init__(self, consumer_key, consumer_secret, redirect_uri): """ 需要提供应用密钥和回调地址 参数获取链接:https://dev.twitter.com/apps redirect_uri必须填写一个可以访问的网址 """ self._consumer_key = consumer_key self._consumer_secret = consumer_secret self._redirect_uri = redirect_uri self._access_token = "" self._access_token_secret = "" self.is_access = False def access(self): """ 获取授权 return True is success """ flag = False self.is_access = False self.auth = tweepy.OAuthHandler(self._consumer_key, self._consumer_secret) logging.info("请求授权URL") try: url = self.auth.get_authorization_url(False,'HMAC-SHA1') logging.info("请求授权URL-成功") except Exception as e: logging.error("请求授权URL-失败") return flag logging.info("进行授权验证") self._form = webForm() self._form.set_callback(self.webview_callback) self._form.show(url, "分享到Twitter", True) if self.is_access == True: flag = True return flag def webview_callback(self, url): if QUrl(self._redirect_uri).host() == url.host(): query = QUrlQuery(url.query()) items = query.queryItems() if self._get_access_token(items) == True: self._form.close() def is_access(self): return self.is_access def _get_access_token(self, list): """ 获取授权token return True is success """ for i in list: if i[0] == 'oauth_verifier': try: self._access_token, self._access_token_secret = self.auth.get_access_token(i[1]) self.api = tweepy.API(self.auth) self._client_id = self.api.me().id self.is_access = True logging.info("请求授权 成功") logging.info("用户名: " + self.api.me().name) #logging.info("access_token: " + self._access_token) #logging.info("access_token_secret: " + self._access_token_secret) return True except Exception as e: self.is_access = False logging.error("请求授 权失败 code:%d\n msg:%s",e.args) #raise e return False def post_tweet(self, text): """ 发布tweet 不要连续发两条一样的推,不然会遇到错误 {'code': 187, 'message': 'Status is a duplicate.'} return True is success """ if self.is_access == False: logging.info("没有获取授权!") return False if text == "": logging.info("tweet文不能为空!") return False try: self.api.update_status(text) logging.info("发布tweet 成功") except Exception as e: logging.error("发布tweet 失败: ",e.args) return False return True def post_tweet_with_media(self,text,filenames): """ 发布带媒体的tweet text = "hello world" filenames = ['C:\\1.png', 'C:\\2.png', ...] return True is success """ if self.is_access == False: logging.info("没有获取授权!") return False try: # upload images and get media_ids media_ids = [] for filename in filenames: res = self.api.media_upload(filename) media_ids.append(res.media_id) # tweet with multiple images self.api.update_status(status=text, media_ids=media_ids) logging.info("发布带图tweet 成功") except Exception as e: logging.error("发布带图tweet 失败: ",e.args) return False def get_last_tweet(self): """ 获取最新一条tweet return NOT "" is success """ text = "" if self.is_access == False: logging.info("没有获取授权!") else: try: tweet = self.api.user_timeline(id = self._client_id, count = 1)[0] logging.info("获取tweet 成功") text = tweet.text except Exception as e: logging.error("获取tweet 失败: ",e.args) return text
测试代码
if __name__ == '__main__': logging.basicConfig(level=logging.INFO) tw = Twitter("Consumer Key","Consumer Secret","https://www.baidu.com") if tw.access() == True: print("验证成功") import random text = "Hello World " + str(random.random()) tw.post_tweet(text) #filenames = ['C:\\1.png', 'C:\\2.png'] filenames = ['output.mp4'] if tw.post_tweet_with_media(text,filenames) == True: print("分享成功") else: print("分享失败")

Youtube

概述

Youtube 上只能分享视频。Youtube API 使用Google开发者后台,所以需要先与Google对接。

需要登录验证只能分享视频需要建立开发者应用Youtube使用Google体系,需要通过google验证需要真实有效,并且通过google所有权验证的回调地址需要多个第三方库需要凭据文件

相关链接

Google API Console https://console.developers.google.com/apisSearch Console https://www.google.com/webmasters/tools/home?hl=zh-CNYoutubeScope https://developers.google.com/youtube/v3/guides/auth/client-side-web-appsAPI索引 https://developers.google.com/youtube/v3/docs/python samples https://developers.google.com/youtube/v3/code_samples/python

环境配置

开发者后台
打开Google API Console https://console.developers.google.com/apis新建项目点开菜单,依次点击API和服务→信息中心→启用API和服务搜索YouTube Data API v3并启用点开菜单,依次点击凭据→OAuth同意屏幕,填写相关资料并保存打开Search Console https://www.google.com/webmasters/tools/home?hl=zh-CN输入回调网址,点击添加属性。按照要求验证所填网址回到Google API Console,点击网域验证→添加网域,填入刚才验证通过的网址点击凭据→创建凭据→OAuth客户端ID,选择网页客户端,填写相关资料并保存下载client_secret.json凭据文件
运行库
pip install --upgrade google-api-python-client pip install --upgrade google-auth google-auth-oauthlib google-auth-httplib2 pip install oauth2client pip install apiclient pip install httplib2
代码
import logging import time import httplib2 import oauth2client.client import oauth2client.file import oauth2client.tools import googleapiclient.discovery import googleapiclient.errors import googleapiclient.http class Youtube(): def __init__(self, client_secret_path, redirect_uri): """ 需要提供凭据文件和回调地址 1.在Google API Console新建应用 (https://console.developers.google.com/apis) 2.生成Oauth凭据 3.新建网页客户端 4.填写登录回调地址redirect_uri,地址必须在 Search Console中认证通过 (https://www.google.com/webmasters/tools) 5.到凭据页面下载网页客户端的凭据client_secret.json (可以考虑把凭据文件放到服务器上) """ self._client_secret_file = client_secret_path self._callback_url = "https://raw.githubusercontent.com/aa13058219642/facebook-callback/master" #请求权限 (其他权限详见 #https://developers.google.com/youtube/v3/guides/auth/client-side-web-apps) self._scopes = ['https://www.googleapis.com/auth/youtube','https://www.googleapis.com/auth/youtube.upload'] self._access_token = "" self.is_access = False def access(self): """ 获取授权 return True is success """ state = False self.is_access = False self.flow = oauth2client.client.flow_from_clientsecrets(self._client_secret_file, self._scopes,redirect_uri=self._callback_url, prompt='consent') url = self._get_authorize_url() if url != None: logging.info("进行Google授权验证") self._form = webForm() self._form.set_callback(self.webview_callback) self._form.show(url,"分享到Youtube") if self.is_access == True: state = True return state def _get_authorize_url(self): """ 请求授权URL return NOT None success """ url = None try: logging.info("请求授权URL") url = self.flow.step1_get_authorize_url() except Exception as e: logging.error("请求授权URL-失败") url = None return url def _get_access_token(self, list): """ 获取授权token return True is success """ state = False for i in list: if i[0] == 'code': try: credentials = self.flow.step2_exchange(i[1]) #保存授权文件 #storage = oauth2client.file.Storage('credentials.json') #storage.put(credentials) http = httplib2.Http() http = credentials.authorize(http) self._youtube = googleapiclient.discovery.build("youtube","v3",http = http,cache_discovery=False) self.is_access = True logging.info("请求授权 成功") #logging.info("用户名: " + self.api.me().name) #logging.info("access_token: " + self._access_token) #logging.info("access_token_secret: " + self._access_token_secret) state = True except Exception as e: self.is_access = False logging.error("请求授 权失败 code:%d\n msg:%s",e.args) elif i[0] == "error": logging.info("请求授权 失败") for j in list: logging.info(j[0] + ": " + j[1]) break return state def webview_callback(self, url): if QUrl(self._callback_url).host() == url.host(): query = QUrlQuery(url.query()) items = query.queryItems() if self._get_access_token(items) == True: self.is_access = True else: self.is_access = False self._form.close() pass def post(self,text): print("post") def upload(self, filename, title, description='', category='22', keywords='', privacyStatus='public'): """ 上传视频 filename: 文件名及路径 title: 视频标题 description: 视频描述 category: 视频类别ID(id列表: https://gist.github.com/dgp/1b24bf2961521bd75d6c) keywords: 关键字 (以英文逗号分隔) privacyStatus: 隐私权限 ('public', 'private', 'unlisted') """ if self.is_access == False: logging.info("没有获取授权!") return False tags = None if keywords: tags = keywords.split(',') body = dict(snippet=dict(title=title, description=description, tags=tags, categoryId=category), status=dict(privacyStatus=privacyStatus)) # Call the API's videos.insert method to create and upload the video. insert_request = self._youtube.videos().insert(part=','.join(body.keys()), body=body, # The chunksize parameter specifies the size of each chunk of data, # in bytes, that will be uploaded at a time. Set a higher value for # reliable connections as fewer chunks lead to faster uploads. # Set a lower value for better recovery on less reliable connections. # # Setting 'chunksize' equal to -1 in the code below means that the entire # file will be uploaded in a single HTTP request. (If the upload fails, # it will still be retried where it left off.) This is usually a best # practice, but if you're using Python older than 2.6 or if you're # running on App Engine, you should set the chunksize to something # like 1024 * 1024 (1 megabyte). media_body=googleapiclient.http.MediaFileUpload(filename, chunksize=-1, resumable=True)) # This method implements an exponential backoff strategy to resume a # failed upload. response = None error = None retry = 0 while response is None: try: logging.info('Uploading file...') status, response = insert_request.next_chunk() if response is not None: if 'id' in response: logging.info('Video id "%s" was successfully uploaded.' % response['id']) else: logging.info('The upload failed with an unexpected response: %s' % response) except Exception as e: error = 'A retriable error occurred: %s' % e if error is not None: logging.info(error) retry += 1 if retry > MAX_RETRIES: logging.info('No longer attempting to retry.') max_sleep = 2 ** retry sleep_seconds = random.random() * max_sleep logging.info('Sleeping %f seconds and then retrying...' % sleep_seconds) time.sleep(sleep_seconds) return True
测试代码
if __name__ == '__main__': logging.basicConfig(level=logging.INFO) youtu = Youtube('client_secret.json',"https://www.example.com") if youtu.access() == True: print("验证成功\n开始上传视频") video = "output.mp4" title = "Hello World" if youtu.upload(video,title)==True: print("分享成功") else: print("分享失败")

Facebook

概述

Facebook停用了public_actions权限以后,就不再支持登录后接管账户的操作了,要通过facebook分享内容只能通过额外提供的分享页面,该页面只能分享网址。不过可以通过在目标网址中添加mate标签来实现部分自定义内容。

需要登录验证只能用户手动发布分享,不能自动执行只能分享链接,但可以部分自定义需要填写回调网址需要应用ID

相关链接

开发者后台 https://developers.facebook.com/appsFacebook 分享文档 https://developers.facebook.com/docs/sharing/web目标网站构建指南 https://developers.facebook.com/docs/sharing/webmasters#markup

环境配置

开发者后台
打开 https://developers.facebook.com/apps新建APP,填写相关资料在左边菜单点击添加产品→facebook登录在左边菜单点击→facebook登录→设置,输入有效 OAuth 跳转 URI和取消授权的回调地址在左边菜单点击设置→基本,填写应用域名及其他相关资料并保存记下应用编号app_id
运行库

Facebook分享不需要第三方库

代码
import logging class Facebook(): def __init__(self, app_id, redirect_uri): """ 需要提供应用ID和回调地址 app_id: 获取步骤详见 https://developers.facebook.com/ redirect_uri: 必须是开发者应用中的“应用域名”属性里的子域名 (https://developers.facebook.com/apps/{{{{app_id}}}}/settings/basic/) """ self._app_id = app_id #self._app_secret = "" self._redirect_uri = redirect_uri self.is_access = False def webview_callback(self, url): if QUrl(self._redirect_uri).host() == url.host(): self.is_access = True return def post(self, link, hashtag=None, display=None): """打开发布动态窗口 link: 要分享的目标网址(目标网站配置指南: https://developers.facebook.com/docs/sharing/webmasters#markup) hashtag: 话题标签(不需要带“#”号) display: 展示方式(详见 https://developers.facebook.com/docs/sharing/reference/share-dialog) """ logging.info("登录验证") if display == None: display = "popup" tag = "" if hashtag == None else "&hashtag=#" + hashtag url = ("https://www.facebook.com/dialog/feed?" + "app_id=" + self._app_id + "&display=" + display + "&link=" + link + "&redirect_uri=" + self._redirect_uri + tag) self.is_access = False self._form = webForm() self._form.set_callback(self.webview_callback) self._form.show(url,"分享到facebook") return self.is_access
测试代码
if __name__ == '__main__': logging.basicConfig(level=logging.INFO) link = "https://www.google.com/" fb = Facebook("app_id","https://www.baidu.com/") if fb.post("https://www.google.com/","hellooo") == True: print("分享成功") else: print("分享失败")
转载请注明原文地址: https://www.6miu.com/read-2350373.html

最新回复(0)