浏览器
概述
第三方登录平台免不了需要登录验证，比起命令行输入账号密码再填充表单，不如直接用浏览器实在
环境配置
运行库
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):
if self._debug ==
True:
query = QUrlQuery(url.query())
items = query.queryItems()
logging.info(
"--------------------------------------")
logging.info(
"[url]: " + url.toString())
logging.info(
"[host]: " + url.host())
for i
in items:
logging.info(
" " + i[
0] +
" : " + i[
1])
if self.callbackFunc !=
None:
self.callbackFunc(url)
pass
概述
使用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)
return True
except Exception
as e:
self.is_access =
False
logging.error(
"请求授 权失败 code:%d\n msg:%s",e.args)
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:
media_ids = []
for filename
in filenames:
res = self.api.media_upload(filename)
media_ids.append(res.media_id)
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 = [
'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"
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])
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(
"请求授权 成功")
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))
insert_request = self._youtube.videos().insert(part=
','.join(body.keys()),
body=body,
media_body=googleapiclient.http.MediaFileUpload(filename, chunksize=-
1, resumable=
True))
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._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(
"分享失败")
