Python:爬虫技术

from urllib import parse

u = parse.urlencode({
    "url":"http://www.xx.com/python",
    "p_url":"http://www.xx.com/python?id=1&name=张三"
})
print(u)

# 运行结果
url=http%3A%2F%2Fwww.xx.com%2Fpython&p_url=http%3A%2F%2Fwww.xx.com%2Fpython%3Fid%3D1%26name%3D%E5%BC%A0%E4%B8%89

# 页面使用utf-8编码
# https://www.baidu.com/s?wd=中
# 上面的url编码后，如下
# https://www.baidu.com/s?wd=%E4%B8%AD

from urllib import parse

u = parse.urlencode({"wd":"中"}) #编码
url= "https://www.baidu.com/s?{}".format(u)
print(url)

print("中".encode("utf-8")) # b'xe4\xb8\xad'
print(parse.unquote(u)) #解码
print(parse.unquote(url))

from urllib.request import Request, urlopen
from urllib import parse

baseurl = 'https://cn.bing.com/search'
params = parse.urlencode({
    'q':'jasper'
})
url = "{}?{}".format(baseurl, params)
print(url)

ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'

request = Request(url, headers={'user-agent':ua})
with urlopen(request) as response:
    # print(response.read())
    with open('d:/tmp/a.html', 'wb') as f:
        f.write(response.read())

from urllib.request import Request, urlopen
from urllib import parse
import simplejson

url = 'https://httpbin.org/post'
params = parse.urlencode({
    'q':'jasper'
})

ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'

request = Request(url, headers={
    'X-REQUEST':'CICI',
    'user-agent':ua
})

with urlopen(request, params.encode('utf-8')) as response:
    print(simplejson.loads(response.read()))

{
    'args': {

    },
    'data': '',
    'files': {

    },
    'form': {
        'q': 'jasper'
    },
    'headers': {
        'Accept-Encoding': 'identity',
        'Content-Length': '8',
        'Content-Type': 'application/x-www-form-urlencoded',
        'Host': 'httpbin.org',
        'User-Agent': 'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,
        likeGecko)Chrome/138.0.0.0Safari/537.36',
        'X-Amzn-Trace-Id': 'Root=1-68932b41-6306d8875659ae8129ecb0ad',
        'X-Request': 'CICI'
    },
    'json': None,
    'origin': '101.71.195.151',
    'url': 'https: //httpbin.org/post'
}

from urllib.request import Request, urlopen
from urllib import parse
import json

url = 'https://httpbin.org/post'
url = 'https://movie.douban.com/j/search_subjects'
params = parse.urlencode({
    "tag":"热门",
    "type": "movie",
    "page_limit": 5,
    "page_start": 0 # 偏移

})

url = "{}?{}".format(url, params)

ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'

# POST
request = Request(url, headers={
    'user-agent':ua
})
with urlopen(request, params.encode('utf-8')) as response:  # urlopen(url, data) data不为None就是POST请求
    print(response.status, response.reason, '++++')
    j = json.loads(response.read())
    print(len(j['subjects']))
print('-'*30)

<?xml version="1.0" encoding="utf-8"?>
<bookstore>
  <book id="bk101">
    <author>Gambardella, Matthew</author>
    <title>XML Developer's Guide</title>
    <genre>Computer</genre>
    <price>44.95</price>
    <publish_date>2000-10-01</publish_date>
    <description>An in-depth look at creating applications with XML.</description>
  </book>
  <book id="bk102" class="bookinfo even">
    <author>Ralls, Kim</author>
    <title>Midnight Rain</title>
    <genre>Fantasy</genre>
    <price>5.95</price>
    <publish_date>2000-12-16</publish_date>
    <description>A former architect battles corporate zombies, an evil sorceress, and her own childhood to become queen of the world.</description>
  </book>
  <book id="bk103">
    <author>Corets, Eva</author>
    <title>Maeve Ascendant</title>
    <genre>Fantasy</genre>
    <price>5.95</price>
    <publish_date>2000-11-17</publish_date>
    <description>After the collapse of a nanotechnology society in England, the young survivors lay the foundation for a new society.</description>
  </book>
  <book id="bk104">
    <author>Corets, Eva</author>
    <title>Oberon's Legacy</title>
    <genre>Fantasy</genre>
    <price>5.95</price>
    <publish_date>2001-03-10</publish_date>
    <description>In post-apocalypse England, the mysterious agent known only as Oberon helps to create a new life for the inhabitants of London. Sequel to Maeve Ascendant.</description>
  </book>
</bookstore>

<div class="billboard-bd">
            <table>
                    <tbody><tr>
                        <td class="order">1</td>
                        <td class="title"><a onclick="moreurl(this, {from:'mv_rk'})" href="https://movie.douban.com/subject/36809864/">南京照相馆</a></td>
                    </tr>
                    <tr>
                        <td class="order">2</td>
                        <td class="title"><a onclick="moreurl(this, {from:'mv_rk'})" href="https://movie.douban.com/subject/36448279/">罗小黑战记2</a></td>
                    </tr>

from lxml import etree
import requests

url = 'https://movie.douban.com/'
ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36'

response = requests.get(url, headers={'user-agent': ua})

with response:
    print(response.status_code)
    if 200 <= response.status_code < 300:
        print(response.text[:300]) # html内容
        root = etree.HTML(response.content) # 分析html,返回DOM根节点
        print(root.xpath('//div[@class="billboard-bd"]//td/a/text()'))

200
<!DOCTYPE html>
<html lang="zh-CN" class="ua-windows ua-webkit">
<head>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8">
    <meta name="renderer" content="webkit">
    <meta name="referrer" content="always">
    <meta name="google-site-verification" content="ok0wCgT20tBBgo9_z
['南京照相馆', '罗小黑战记2', '戏台', '关于约会的一切', '冥婚红包', '长安的荔枝', '花漾少女杀人事件', '没事，没事，没事！', '麦克白', '邓南遮占领阜姆自由邦']

./bin/mongosh "mongodb://localhost:27017"
#旧版本 bin/mongo.exe

#操作
help 打开帮助
show dbs    查看当前库
use blog    有就切换过去，没有就创建后切换过去。刚创建的并不在数据库列表中，需要写入数据后才能看到
db          查看当前数据库
db.users.insert({user:"tom",age:20}) db指代当前数据库；users集合名

test> help
  Shell Help:

    log                                        'log.info(<msg>)': Write a custom info/warn/error/fatal/debug message to the log file
                                               'log.getPath()': Gets a path to the current log file

    use                                        Set current database
    show                                       'show databases'/'show dbs': Print a list of all available databases
                                               'show collections'/'show tables': Print a list of all collections for current database
                                               'show profile': Prints system.profile information
                                               'show users': Print a list of all users for current database
                                               'show roles': Print a list of all roles for current database
                                               'show log <name>': Display log for current connection, if name is not set uses 'global'
                                               'show logs': Print all logger names.
    exit                                       Quit the MongoDB shell with exit/exit()/.exit
    quit                                       Quit the MongoDB shell with quit/quit()
    Mongo                                      Create a new connection and return the Mongo object. Usage: new Mongo(URI, options [optional])
    connect                                    Create a new connection and return the Database object. Usage: connect(URI, username [optional], password [optional])
    it                                         result of the last line evaluated; use to further iterate
    version                                    Shell version
    load                                       Loads and runs a JavaScript file into the current shell environment
    enableTelemetry                            Enables collection of anonymous usage data to improve the mongosh CLI
    disableTelemetry                           Disables collection of anonymous usage data to improve the mongosh CLI
    passwordPrompt                             Prompts the user for a password
    sleep                                      Sleep for the specified number of milliseconds
    print                                      Prints the contents of an object to the output
    printjson                                  Alias for print()
    convertShardKeyToHashed                    Returns the hashed value for the input using the same hashing function as a hashed index.
    cls                                        Clears the screen like console.clear()
    isInteractive                              Returns whether the shell will enter or has entered interactive mode

  For more information on usage: https://mongodb.com/docs/manual/reference/method
test> show dbs
admin    40.00 KiB
config  108.00 KiB
local    40.00 KiB
test> db                                                                                                                                                                
test                                                                                                                                                                    
test> use aaa      # 没有写数据的话，退出后不会创建库。                                                                                                                                                     
switched to db aaa                                                                                                                                                      
aaa>

aaa> use users                                                                                                                                                          
switched to db users                                                                                                                                                    
users> db.users.insert({'name':'tom', 'age':20})  #插入数据                                                                                             
DeprecationWarning: Collection.insert() is deprecated. Use insertOne, insertMany, or bulkWrite.                                                                         
{                                                                                                                                                                       
  acknowledged: true,                                                                                                                                                   
  insertedIds: { '0': ObjectId('6895bc37ad8daf59bceec4a9') }                                                                                                            
}                                                                                                                                                                       
users> db.users.find()                                                                                                                                                  
[ { _id: ObjectId('6895bc37ad8daf59bceec4a9'), name: 'tom', age: 20 } ]                                                                                                 
users> 

#pymongo-4.14.0
pip install pymongo

from pymongo import MongoClient
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
print(client)

db:Database = client['users']
# db = client.users # 与上面等价，因为MongoClient有__getitem__ __getattr__ 方法
print(type(db), db)

table = db['users']
print(type(table), table)

result = table.find() # select * from users.users
print(type(result), *result)
client.close()

# with client:

MongoClient(host=['127.0.0.1:27017'], document_class=dict, tz_aware=False, connect=True)
<class 'pymongo.synchronous.database.Database'> Database(MongoClient(host=['127.0.0.1:27017'], document_class=dict, tz_aware=False, connect=True), 'users')
<class 'pymongo.synchronous.collection.Collection'> Collection(Database(MongoClient(host=['127.0.0.1:27017'], document_class=dict, tz_aware=False, connect=True), 'users'), 'users')
<class 'pymongo.synchronous.cursor.Cursor'> {'_id': ObjectId('6895bc37ad8daf59bceec4a9'), 'name': 'tom', 'age': 20}

from pymongo import MongoClient
from pymongo.collection import Collection
from bson.objectid import ObjectId

client = MongoClient("mongodb://127.0.0.1:27017")
db = client["blog"] #指定数据库
users:Collection = db.users #集合

# 查询
result = users.find_one({"name":"tom"})
print(type(result),result)

# 使用key查询
result2 = users.find_one({"_id":ObjectId("5d539433c1b3dcb23654814d")})
print(type(result2),result2)

#查不到，返回None
result3 = users.find_one({"name":"tommy"})
print(type(result3),result3)1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.

from pymongo import MongoClient
from pymongo.collection import Collection

client = MongoClient("mongodb://127.0.0.1:27017")
db = client["blog"] #指定数据库
users:Collection = db.users #集合

# 多条查询
results = users.find({"name":"tom"})
print(type(results))
print(results)

print("- "*30)
for x in results:
    print(type(x),x)

from pymongo import MongoClient
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
db:Database = client['users']
table = db['users'] # db.users

# 操作符
# result = table.find({'age':{"$gt":20}}) # select * from users.users where name='tom' and age > 20
# result = table.find({'age':{'$nin':[20, 16]}})

# result = table.find({'$and':[
#     {'name':'tom'}, {'age':{'$gt':20}}
# ]})
# result = table.find({'name':'tom', 'age':{'$gt':20}}) #与上面等价

# result = table.find({'$or':[
#     {'age':{'$lt':20}}, {'age':{'$gt':40}}
# ]})
# result = table.find({'name':{'$not':{'$eq':'tom'}}})

# result = table.find({'id':{'$type':16}})
# result = table.find({'Name':{'$exists':True}})

# result = table.find({'name':{'$regex':'^t'}})
result = table.find({'age':{'$mod':[10, 2]}})
print(*result, sep='\n')

client.close()

from pymongo import MongoClient
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
db:Database = client['users']
table = db['users'] # db.users

# 投影
# result = table.find({'age':{'$gt':20}}, ('name',))
# result = table.find({'age':{'$gt':20}}, ['name', 'id']) #允许
result = table.find({'age':{'$gt':20}}, {'name':0, 'age':False}) #排除。 0为排除，非0为允许。但不能混写。
for x in result:
    print(type(x), x)

client.close()

from pymongo import MongoClient
from pymongo.collection import Collection

client = MongoClient("mongodb://127.0.0.1:27017")
db = client["blog"] #指定数据库
users:Collection = db.users #集合

# 统计
# 被弃用
# count = users.find({"age":{"$gt":10}}).count()

#求出age>10的数量
count = users.count_documents({"age":{"$gt":10}})
print(count) #返回结果41.2.3.4.5.6.7.8.9.10.11.12.13.14.

result = table.find({'age':{'$gt':20}}).distinct('name') #去 重

from pymongo import MongoClient
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
db:Database = client['users']
table = db['users'] # db.users

# 聚合
result = table.aggregate([
    {'$match':{'age':{'$gte':20}}}, #过滤
    { '$count':'age'}
])
print(*result)
result = table.count_documents({'age':{'$gte':20}})
print(result)
client.close()

from pymongo import MongoClient
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
db:Database = client['users']
table = db['users'] # db.users

# 聚合
result = table.aggregate([
    {'$match':{'age':{'$gte':5}}},
    {'$group':{'_id':'$name', 'res':{'$sum':'$age'}}} # 注意前面加$。 按name分组，res为age合
])
print(*result)

client.close()

from pymongo import MongoClient, ASCENDING, DESCENDING
from pymongo.synchronous.database import Database

url = 'mongodb://127.0.0.1:27017'
client = MongoClient(url)
db:Database = client['users']
table = db['users'] # db.users

# 排序
result = table.find().sort('age', DESCENDING)
print(*list(result), sep='\n')
print('=' * 30)

result2 = table.find().sort([
    ('name', DESCENDING), # name降序
    ('age', ASCENDING) # age升序
])
print(*result2, sep='\n')

client.close()

from pymongo import MongoClient
from pymongo.collection import Collection

client = MongoClient("mongodb://127.0.0.1:27017")
db = client["blog"] #指定数据库
users:Collection = db.users #集合

# 分页
results = users.find()
print(*list(results),sep="\n")
print("- "*30)

results1 = users.find().skip(2) #跳过前2个
print(*list(results1),sep="\n")
print("- "*30)

results2 = users.find().skip(1).limit(2) #跳过前1个最多显示2个
print(*list(results2),sep="\n")
print("- "*30)

pip install bs4
#pip install beautifulsoup4

pip install lxml

from bs4 import BeautifulSoup

#文件对象
soup = BeautifulSoup(open("test.html"))
# 标记字符串
soup = BeautifulSoup("<html>data</html>")

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>首页</title>
</head>
<body>
<h1>TEST</h1>
<div id="main">
    <h3 class="title highlight"><a href="http://www.python.org">python</a>高级班</h3>
    <div class="content">
        <p id="first">字典</p>
        <p id="second">列表</p>
        <input type="hidden" name="_csrf" value="absdoia23lkso234r23oslfn">
        <!-- comment -->
        <img id="bg1" src="http://www.cici.com/">
        <img id="bg2" src="http://httpbin.org/">
    </div>
</div>
<p>bottom</p>
</body>

from bs4 import BeautifulSoup

with open('test.html', encoding='utf-8') as f:
    doc = BeautifulSoup(f, 'lxml')
    print(doc.prettify())

find_all(name=None,attrs={},recursive=True,string=None,limit=None,**kwargs) #立即返回一个列表

from bs4 import BeautifulSoup

with open("test.html",encoding="utf-8") as f:
    soup = BeautifulSoup(f,"lxml")
    #元素选择器
    print(1,soup.select("p")) #所有的p标签

    #类选择器
    print(2,soup.select(".title"))

    #使用了伪类
    #直接子标签是p的同类型的所有p标签中的第二个
    #(同类型)同标签名p的第2个，nth-of-type，且要求是数字
    #最新版本实现更多伪类支持
    #参考soupsieve.css_parser.CSSParser.parse_pseudo_class
    print(3,soup.select("div.content >p:nth-of-type(2)"))

    # id选择器
    print(4,soup.select("p#second"))
    print(5,soup.select("#bg1"))

    #后代选择器
    print(6,soup.select("div p")) # div下逐层找p
    print(7,soup.select("div div p")) #div下逐层找div下逐层找p

    #子选择器，直接后代
    print(8,soup.select("div > p")) #div下直接子标签的p，有2个

    #相邻兄弟选择器
    print(9, soup.select("div p:nth-of-type(1) + [src]")) #返回[]
    print(9, soup.select("div p:nth-of-type(1) + p"))  # 返回p标签
    print(9, soup.select("div > p:nth-of-type(2) + input"))  # 返回input Tag
    print(9, soup.select("div > p:nth-of-type(2) + [type]"))  # 同上

    #普通兄弟选择器
    print(10, soup.select("div p:nth-of-type(1) ~ [src]")) #返回2个img

    #属性选择器
    print(11,soup.select("[src]")) #有属性src
    print(12,soup.select("[src='/']")) #属性src等于/
    print(13,soup.select("[src='http://www.cici.com/']")) #完全匹配
    print(14,soup.select("[src^='http://www']")) #以http://www开头
    print(15,soup.select("[src$='com/']")) #以com/结尾
    print(16,soup.select("img[src*='cici']")) #包含cici
    print(17,soup.select("img[src*='.com']")) #包含.com
    print(18,soup.select("[class='title highlight']")) #完全匹配calss等于'title highlight'
    print(19,soup.select("[class~=title]")) #多值属性中有一个title

from bs4 import BeautifulSoup
import re

with open('test.html', encoding='utf-8') as f:
    doc = BeautifulSoup(f, 'lxml')
    # html解析3种方法 lxml xpath; bs4 find*; bs4 select css selector
    # 1 标签选择器
    print(doc.select('p')) # 相当于写xpath
    # 2 id
    print(doc.select('p#first'))
    # 3 group
    print(doc.select('p#first,p#second'))
    # 4 class
    print(doc.select('.title'))
    # 5 后代 descendants
    print(doc.select('div h3'))
    print(doc.select('div > h3'))
    # 6 伪类 pseudo-class
    print(doc.select('div.content > p:nth-of-type(2)'))
    # 7 sibling
    # + 直接兄弟
    print(doc.select('div.content > p:nth-of-type(2) + input'))
    # ~ 后续兄弟
    print(doc.select('div.content > p:nth-of-type(2) ~ input'))
    print(doc.select('div.content > p:nth-of-type(2) ~ img'))

    # 8 属性 attribute
    print(doc.select('[src],a[href]')) # 匹配src属性或者a标签中具有href属性的
    print(doc.select('[id=bg2]'))
    print(doc.select('[class="title highlight"]'))
    print(doc.select('[class~="title"]')) # 匹配多值属性中的一个
    print(doc.select('[class*="title"]')) # 包含title
    print(doc.select('[class^="title"]')) # 前面有title
    print(doc.select('[class$="highlight"]')) # 后面有highlight

from bs4 import BeautifulSoup

with open("test.html",encoding="utf-8") as f:
    soup = BeautifulSoup(f,"lxml")
    # 元素选择器
    ele = soup.select("div") #所有的div标签
    print(type(ele))
    print(ele[0].string) #内容仅仅只能是文本类型，否则返回None
    print(list(ele[0].strings)) #迭代保留空白字符
    print(list(ele[0].stripped_strings)) #迭代不保留空白字符

    print("- "*30)
    print(ele[0])
    print("- " * 30)

    print(list(ele[0].text))#本质上就是get_text(),保留空白字符的strings
    print(list(ele[0].get_text())) #迭代并join，保留空白字符，strip默认为False
    print(list(ele[0].get_text(strip=True))) #迭代并join，不保留空白字符

from bs4 import BeautifulSoup
import re

with open('test.html', encoding='utf-8') as f:
    doc = BeautifulSoup(f, 'lxml')

    print(doc.find('div').string)
    print(*doc.find('div').strings)
    print(doc.find('div').text)
    print(*doc.find('div').stripped_strings, sep=',')
    print(doc.find('div').get_text(separator=',' , strip=True))

class Tag(PageElement):
@property
    def string(self):
        if len(self.contents) != 1:
            return None
        child = self.contents[0]
        if isinstance(child, NavigableString):
            return child
        return child.string

    @string.setter
    def string(self, string):
        self.clear()
        self.append(string.__class__(string))

    def _all_strings(self, strip=False, types=(NavigableString, CData)):
        for descendant in self.descendants:
            if (
                (types is None and not isinstance(descendant, NavigableString))
                or
                (types is not None and type(descendant) not in types)):
                continue
            if strip:
                descendant = descendant.strip()
                if len(descendant) == 0:
                    continue
            yield descendant

    strings = property(_all_strings)

    @property
    def stripped_strings(self):
        for string in self._all_strings(True):
            yield string

    def get_text(self, separator="", strip=False,
                 types=(NavigableString, CData)):
        return separator.join([s for s in self._all_strings(
                    strip, types=types)])
    getText = get_text
    text = property(get_text)

yum -y install erlang-21.3.8.6-1.el7.x86_64.rpm
rabbitmq-server-3.7.16-1.el7.noarch.rpm

[root@xdd ~]# rpm -ql rabbitmq-server 
/etc/logrotate.d/rabbitmq-server
/etc/profile.d/rabbitmqctl-autocomplete.sh
/etc/rabbitmq
/usr/lib/ocf/resource.d/rabbitmq/rabbitmq-server
/usr/lib/ocf/resource.d/rabbitmq/rabbitmq-server-ha
/usr/lib/rabbitmq/autocomplete/bash_autocomplete.sh
/usr/lib/rabbitmq/autocomplete/zsh_autocomplete.sh
/usr/lib/rabbitmq/bin/cuttlefish
/usr/lib/rabbitmq/bin/rabbitmq-defaults

rabbitmq-plugins list

rabbitmq-plugins enable rabbitmq_management

systemctl start rabbitmq-server

[root@xdd ~]# ss -tanl | grep 5672
LISTEN     0      128          *:25672                    *:*
LISTEN     0      128          *:15672                    *:*
LISTEN     0      128         :::5672                    :::*
[root@xdd ~]#

└─$ docker exec -i rabbitmq  rabbitmqctl          
rabbitmqctl [--node <node>] [--timeout <timeout>] [--longnames] [--quiet] <command> [<command options>]
Users:
   add_user                                      添加用户
   authenticate_user                             Attempts to authenticate a user. Exits with a non-zero code if authentication fails.
   change_password                               修改用户名，密码
   clear_password                                Clears (resets) password and disables password login for a user
   clear_user_limits                             Clears user connection/channel limits
   delete_user                                   删除用户
   list_user_limits                              Displays configured user limits
   list_users                                    列出用户
   set_user_limits                               Sets user limits
   set_user_tags                                 设置用户tag

docker exec -i rabbitmq  rabbitmqctl  help add_user

rabbitmqctl add_user jasper 123456  #添加jasper用户
rabbitmqctl list_users #查看所有用户
rabbitmqctl set_user_tags jasper administrator #设置admin用户为管理员用户

pip install pika

# send.py
import pika
from pika.adapters.blocking_connection import BlockingChannel

#构建用户名密码对象
credential = pika.PlainCredentials("jasper","123456")
# 配置链接参数
params = pika.ConnectionParameters(
    "192.168.61.108",#ip地址
    5672,  #端口
    "test",#虚拟机
    credential #用户名密码
)

# # 第二种建立连接方式
# params = pika.URLParameters("amqp://jasper:[email protected]:5672/test")

# 建立连接
connection = pika.BlockingConnection(params)

with connection:
    # 建立通道
    channel:BlockingChannel = connection.channel()

    #创建一个队列，queue命名为hello，如果queue不存在，消息将被dropped
    channel.queue_declare(queue="hello")

    channel.basic_publish(
        exchange="",#使用缺省exchange
        routing_key="hello", #routing_key必须指定，这里要求和目标queue一致
        body="Hello world" #消息
    )
    print("消息发送成功Sent Message OK")

pika.exceptions.ProbableAuthenticationError: (403, 'ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile.')

# amqp://username:password@host:port/<virtual_host>[?query-string] 
parameters = pika.URLParameters('amqp://guest:guest@rabbit-server1:5672/%2F') 
# %2F指代/，就是缺省虚拟主机

# send.py
import pika
from pika.adapters.blocking_connection import BlockingChannel
import time

# 第二种建立连接方式
params = pika.URLParameters("amqp://jasper:[email protected]:5672/test")
# 建立连接
connection = pika.BlockingConnection(params)

with connection:
    # 建立通道
    channel:BlockingChannel = connection.channel()

    #创建一个队列，queue命名为hello，如果queue不存在，消息将被dropped
    channel.queue_declare(queue="hello")

    for i in range(40):

        channel.basic_publish(
            exchange="",#使用缺省exchange
            routing_key="hello", #routing_key必须指定，这里要求和目标queue一致
            body="data{:02}".format(i) #消息
        )
        time.sleep(0.5)
    print("消息发送成功Sent Message OK")

import pika
import time
#from pika.credentials import PlainCredentials
# sender
# param = pika.ConnectionParameters(
#     '192.168.226.130', '5672',
#     'test', #虚拟机
#     PlainCredentials('jasper', '123456')
# )
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

# 建立连接
connection = pika.BlockingConnection(param)
channel = connection.channel() # 建立通道

channel.queue_declare(queue='hello')
# 生产端声明。如果不存在，立即创建该q。创建的q不会因为连接中断而删除

for i in range(40):
    channel.basic_publish(exchange='', # 不指定，默认缺省 test/default
                          routing_key='hello', # q的名字
                          body='data-{:2}'.format(i))
    time.sleep(1)
print(" [x] Sent 'Hello World!'")

connection.close()

# receie.py
import pika
from pika.adapters.blocking_connection import BlockingChannel

# 建立连接
params = pika.URLParameters("amqp://jasper:[email protected]:5672/test")
connection = pika.BlockingConnection(params)

with connection:
    channel:BlockingChannel = connection.channel()
    msg = channel.basic_get("hello",True) #从名称为hello的queue队列中获取消息，获取不到阻塞
    method,props,body = msg #拿不到的消息tuple为(None,None,None)
    if body:
        print("获取到了一个消息Get A message = {}".format(body))
    else:
        print("没有获取到消息empty")

(<Basic.GetOk(['delivery_tag=1', 'exchange=', 'message_count=38', 'redelivered=False', 'routing_key=hello'])>, <BasicProperties>, b'data01')  
返回元组：(方法method,属性properties,消息body)
无数据返回：(None,None,None)

# receie.py 消费代码
import pika
from pika.adapters.blocking_connection import BlockingChannel

# 建立连接
params = pika.URLParameters("amqp://jasper:[email protected]:5672/test")
connection = pika.BlockingConnection(params)

def callback(channel,method,properties,body):
    print("Get a message = {}".format(body))

with connection:
    channel:BlockingChannel = connection.channel()
    channel.basic_consume(
        "hello",#队列名
        callback,#消费回调函数
        True,#不回应
    )
    print("等待消息，退出按CTRL+C;Waiting for messages. To exit press CTRL+C")
    channel.start_consuming()

import pika
# receive
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

connection = pika.BlockingConnection(param)
channel = connection.channel()

#channel.queue_declare(queue='hello') # q生产端创建的，我们直接拿就行了
def callback(ch, method, properties, body):
    print(f" [x] Received {body}")

channel.basic_consume(queue='hello',
                      auto_ack=True,
                      on_message_callback=callback)

channel.start_consuming()  # 阻塞

import pika
# receive 消费者代码
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')
connection = pika.BlockingConnection(param)

def callback(ch, method, properties, body):
    print(f" [x] Received {body}")
def callback1(ch, method, properties, body):
    print(f" [y] Received {body}")

with connection:
    channel = connection.channel()
    channel.queue_declare(queue='hello')  # 生产端和消费端都要看看q是否存在，不存在就创建。
    # 消费者，每一个消费者使用一个basic_consume
    channel.basic_consume(queue='hello',
                          auto_ack=True,
                          on_message_callback=callback)
    channel.basic_consume(queue='hello',
                          auto_ack=True,
                          on_message_callback=callback1)
    print('Waiting for message. To exit press CTRL+C')
    channel.start_consuming()  # 阻塞

# 生成一个交换机
channel.exchange_declare(
    exchange="logs", #新交换机
    exchange_type="fanout" #广播
)

# 消费者端
result =channel.queue_declare(queue="") #生成一个随机名称的queue
result = channel.queue_declare(queue="",exclusive=True) #生成一个随机名称的queue,并在断开链接时删除queue
# exclusive意思是只允许当前connection访问，当前连接断开时删除queue

# 生成queue
q1:Method = channel.queue_declare(queue="",exclusive=True)
q2:Method = channel.queue_declare(queue="",exclusive=True)
q1name = q1.method.queue #可以通过result.method.queue 查看随机名称
q2name = q2.method.queue
print(q1name,q2name)

#绑定
channel.queue_bind(exchange="logs",queue=q1name)
channel.queue_bind(exchange="logs",queue=q2name)

# send.py 生产者代码
import pika
import time

# 建立连接
params = pika.URLParameters("amqp://jasper:[email protected]:5672/test")
connection = pika.BlockingConnection(params)
channel = connection.channel()

#指定交换机和模式
exchange_name = 'logs'
channel.exchange_declare(
    exchange=exchange_name, #新交换机
    exchange_type="fanout" #扇出，广播
)

with connection:
    for i in range(40):
        channel.basic_publish(
            exchange=exchange_name, #使用指定的exhcange
            routing_key="", #广播模式，不指定routing_key
            body = "data-{:02}".format(i) #消息
        )
        time.sleep(0.01)
    print("Send OK")

# receie.py 消费者代码

import time
import pika

exchange_name = 'logs'
connection = pika.BlockingConnection(pika.URLParameters("amqp://jasper:[email protected]:5672/test"))
channel = connection.channel()
# 指定交换机
channel.exchange_declare(
    exchange=exchange_name, #新交换机
    exchange_type="fanout" #扇出，广播
)

# 生成队列，名称随机，exclusive=True断开删除队列
q1 = channel.queue_declare(queue="",exclusive=True)
q2 = channel.queue_declare(queue="",exclusive=True)
name1 = q1.method.queue #查看队列名
name2 = q2.method.queue

# 绑定到交换机
channel.queue_bind(exchange=exchange_name, queue=name1)
channel.queue_bind(exchange=exchange_name, queue=name2)

def callback(channel,method,properties,body):
    print("{}\n{}".format(channel,method))
    print("Get a message = {}".format(body))

with connection:
    # 消费者，每一个消费者使用一个basic_consume
    channel.basic_consume(queue=name1,
                    auto_ack=True,
                    on_message_callback=callback)
    channel.basic_consume(queue=name2,
                    auto_ack=True,
                    on_message_callback=callback)

    print("Waiting for messages. To exit press CTRL+C")
    channel.start_consuming()

q1 = channel.queue_declare(queue="",exclusive=True)
q2 = channel.queue_declare(queue="",exclusive=True)

import pika
import time
#from pika.credentials import PlainCredentials
# sender
# param = pika.ConnectionParameters(
#     '192.168.226.130', '5672',
#     'test', #虚拟机
#     PlainCredentials('jasper', '123456')
# )
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

# 建立连接
connection = pika.BlockingConnection(param)
channel = connection.channel() # 建立通道

ex_name = 'logs'
ex_type = 'fanout'
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type)

# channel.queue_declare(queue='hello')
# 生产端声明。如果不存在，立即创建该q。创建的q不会因为连接中断而删除

# 消息持久化 1. 队列必须持久化 2.改变消息的属性值
for i in range(50):
    channel.basic_publish(exchange=ex_name, # 不指定，默认缺省 test/default
                          routing_key='', # q的名字
                          # properties=pika.BasicProperties(delivery_mode=2),
                          # 持久化数据，队列必须持久化
                          body='data-{:2}'.format(i)
    time.sleep(0.5)
print(" [x] Sent 'Hello World!'")

connection.close()

import pika
import time
# receive
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

connection = pika.BlockingConnection(param)
channel = connection.channel()

ex_name = 'logs'
ex_type = 'fanout'
# 生成交换机
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type)
result = channel.queue_declare(queue='', exclusive=True)
q_name1 = result.method.queue
channel.queue_bind(exchange=ex_name, queue=q_name1)
result2 = channel.queue_declare(queue='', exclusive=True)
q_name2 = result2.method.queue
channel.queue_bind(exchange=ex_name, queue=q_name2)

def callback(ch, method, properties, body):
    ctag = method.consumer_tag
    dtag = method.delivery_tag
    # time.sleep(5)
    print(f"{ctag} {dtag} ** {body}")
    channel.basic_ack(dtag) # 手动应答

# 返回值为消费者号
# channel.basic_qos(prefetch_count=1)
c1 = channel.basic_consume(queue=q_name1,
                      auto_ack=False,
                      on_message_callback=callback)
c2 = channel.basic_consume(queue=q_name2,
                      auto_ack=False,
                      on_message_callback=callback)
channel.start_consuming()  # 阻塞

import pika
import time
import random
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

# 建立连接
connection = pika.BlockingConnection(param)
channel = connection.channel() # 建立通道

# 交换机
ex_name = 'colors'
ex_type = 'direct'
colors = ('orange', 'black', 'green')
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type) # 路由

# channel.queue_declare(queue='hello')
# 生产端声明。如果不存在，立即创建该q。创建的q不会因为连接中断而删除

# 消息持久化 1. 队列必须持久化 2.改变消息的属性值
for i in range(50):
    rk = random.choice(colors)
    channel.basic_publish(exchange=ex_name, # 不指定，默认缺省 test/default
                          routing_key=rk, # q的名字
                          # properties=pika.BasicProperties(delivery_mode=2),
                          # 持久化数据，队列必须持久化
                          body='{}-data-{:2}'.format(rk, i))
    time.sleep(0.5)
print(" [x] Sent 'Hello World!'")

connection.close()

import pika
import time
# receive
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

connection = pika.BlockingConnection(param)
channel = connection.channel()

ex_name = 'colors'
ex_type = 'direct'
colors = ('orange', 'black', 'green')
# 生成交换机
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type)

# 生成队列，名称随机，exclusive=True断开删除该队列
result = channel.queue_declare(queue='', exclusive=True)
q_name1 = result.method.queue
# 绑定到交换机，而且一定要绑定routing_key
channel.queue_bind(exchange=ex_name, queue=q_name1, routing_key=colors[0])
# channel.queue_bind(q_name1, ex_name, colors[0])

result2 = channel.queue_declare(queue='', exclusive=True)
q_name2 = result2.method.queue
channel.queue_bind(q_name2, ex_name, colors[1])
channel.queue_bind(q_name2, ex_name, colors[2])

def callback(ch, method, properties, body):
    ctag = method.consumer_tag
    dtag = method.delivery_tag
    # time.sleep(5)
    print(f"{ctag} {dtag} ** {body}")
    # channel.basic_ack(dtag) # 手动应答

# 返回值为消费者号
# channel.basic_qos(prefetch_count=1)
c1 = channel.basic_consume(queue=q_name1,
                      auto_ack=True, # 自动应答
                      on_message_callback=callback)
c2 = channel.basic_consume(queue=q_name2,
                      auto_ack=True,
                      on_message_callback=callback)
channel.start_consuming()  # 阻塞
connection.close()

import pika
import time
import random
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

# 建立连接
connection = pika.BlockingConnection(param)
channel = connection.channel() # 建立通道

ex_name = 'products'
ex_type = 'topic'
products = ('pc', 'phone', 'tv')
colors = ('orange', 'black', 'red')
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type)

# channel.queue_declare(queue='hello')
# 生产端声明。如果不存在，立即创建该q。创建的q不会因为连接中断而删除

# 消息持久化 1. 队列必须持久化 2.改变消息的属性值
for i in range(50):
    rk = "{}.{}".format(random.choice(products), random.choice(colors))
    msg = '{}-data-{:2}'.format(rk, i)
    channel.basic_publish(exchange=ex_name, # 不指定，默认缺省 test/default
                          routing_key=rk, # q的名字
                          # properties=pika.BasicProperties(delivery_mode=2),
                          # 持久化数据，队列必须持久化
                          body=msg)
    print(msg)
    time.sleep(0.5)
print(" [x] Sent 'Hello World!'")

connection.close()

import pika
import time
# receive
param = pika.URLParameters('amqp://jasper:[email protected]:5672/test')

connection = pika.BlockingConnection(param)
channel = connection.channel()

ex_name = 'products'
ex_type = 'topic'
products = ('pc', 'phone', 'tv')
colors = ('orange', 'black', 'red')
channel.exchange_declare(exchange=ex_name,
                         exchange_type=ex_type)

result = channel.queue_declare(queue='', exclusive=True)
q_name1 = result.method.queue
channel.queue_bind(q_name1, ex_name, 'phone.*')

result2 = channel.queue_declare(queue='', exclusive=True)
q_name2 = result2.method.queue
channel.queue_bind(q_name2, ex_name, '*.red')

def callback(ch, method, properties, body):
    ctag = method.consumer_tag
    dtag = method.delivery_tag
    # time.sleep(5)
    print(f"{ctag} {dtag} ** {body}")
    # channel.basic_ack(dtag) # 手动应答

# 返回值为消费者号
# channel.basic_qos(prefetch_count=1)
c1 = channel.basic_consume(queue=q_name1,
                      auto_ack=True, # 自动应答
                      on_message_callback=callback)
c2 = channel.basic_consume(queue=q_name2,
                      auto_ack=True,
                      on_message_callback=callback)
channel.start_consuming()  # 阻塞
connection.close()

import requests
from lxml import etree

BASE_URL = 'https://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
headers = {
  'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
}
url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, 1)
print(url)

response = requests.get(url, headers=headers)

with response:
    print(response.status_code)
    if response.status_code == 200:
        xpath = '//h2[@class="news_entry"]/a/text()'
        doc = etree.HTML(response.text)
        e = doc.xpath(xpath)
        print(len(e), e)

import requests
from lxml import etree
from bs4 import BeautifulSoup

BASE_URL = 'https://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
headers = {
  'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
}
url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, 1)
print(url)

response = requests.get(url, headers=headers)

with response:
    print(response.status_code)
    if response.status_code == 200:
        ## xpath方法
        # xpath = '//h2[@class="news_entry"]/a/text()'
        # doc = etree.HTML(response.text)
        # e = doc.xpath(xpath)
        # print(len(e), e)

        doc = BeautifulSoup(response.content, 'lxml')
        # print(doc)
        ## beatifulsoup方法
        ## 方法1 css选择器
        # selector = 'h2.news_entry > a'
        # titles = doc.select(selector)
        # for title in titles:
        #     print(title.get('href'), title.text)

        ## 方法2 find_all
        h2s = doc.find_all('h2', class_='news_entry')
        for h in h2s:
            a = h.a # h.find('a')
            print(a.get('href'), a.text)

import requests
from lxml import etree
from bs4 import BeautifulSoup

BASE_URL = 'https://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
headers = {
  'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
}

urls = [] # 待爬取url,先进先出

# 创建博客园的新闻urls，每页30条新闻
def start_urls(start=1, stop=2, step=1):
    for i in range(start, stop, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        urls.append(url)
    print('链接创建完成')

def crawler():
    for url in urls:
        response = requests.get(url, headers=headers)
        with response:
            print(response.status_code)
            if response.status_code == 200:
                # xpath = '//h2[@class="news_entry"]/a/text()'
                # selector = 'h2.news_entry > a'
                # find_all
                doc = BeautifulSoup(response.content, 'lxml')
                h2s = doc.find_all('h2', class_='news_entry')
                for h in h2s:
                    a = h.a # h.find('a')
                    print(a.get('href'), a.text)

start_urls(1, 2)
print(urls)
crawler()

import requests
from bs4 import BeautifulSoup
from bs4.element import Tag
from queue import Queue
from concurrent.futures import ThreadPoolExecutor


BASE_URL = 'http://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
# https://news.cnblogs.com/n/page/2/ 列表页
# https://news.cnblogs.com/n/628919/ 详情页

headers = {
    'User-agent': "Mozilla/5.0 (Windows; U; Windows NT 6.1; zh-CN) AppleWebKit/537.36 (KHTML,like Gecko)"
                  " Version / 5.0.1Safari / 537.36"
}

# 异步，队列，以后换成第三方队列
urls = Queue()


# 创建博客园的新闻urls，每页30条新闻
def starts_url(start, stop, step=1):
    for i in range(start, stop + 1, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        print(url)
        urls.put(url)  # 加入队列
    print('任务链接创建完毕')


# 爬取页面
def crawler():
    url = urls.get()  # 阻塞，拿一条
    with requests.get(url, headers=headers) as response:
        html = response.text

        # 解析
        soup = BeautifulSoup(html, 'lxml')
        # h2.news_entry > a
        # //h2[@new_entry=""]/a
        titles = soup.select('h2.news_entry > a')
        for title in titles:
            print(title.get('href'), title.text)


# starts_url(1, 1)
# crawler()

# 线程池
executor = ThreadPoolExecutor(10)

executor.submit(starts_url, 1, 1)
for i in range(5):
    executor.submit(crawler)

import requests
from bs4 import BeautifulSoup
from bs4.element import Tag
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
from threading import Event


BASE_URL = 'http://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
# https://news.cnblogs.com/n/page/2/ 列表页
# https://news.cnblogs.com/n/628919/ 详情页

headers = {
    'User-agent': "Mozilla/5.0 (Windows; U; Windows NT 6.1; zh-CN) AppleWebKit/537.36 (KHTML,like Gecko)"
                  " Version / 5.0.1Safari / 537.36"
}

# 异步，队列，以后换成第三方队列
urls = Queue()# 待爬取队列
htmls = Queue() # 待分析队列
outputs = Queue() # 待输出队列



# 创建博客园的新闻urls，每页30条新闻
def starts_urls(start, stop, step=1):
    for i in range(start, stop + 1, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        print(url)
        urls.put(url)  # 加入队列
    print('任务链接创建完毕')


# 爬取页面
def crawler(e:Event):
    while not e.is_set():
        url = urls.get()  # 阻塞，拿一条
        with requests.get(url, headers=headers) as response:
            html = response.text
            htmls.put(html)

# 解析页面
def parse(e:Event):
    # 解析
    while not e.is_set():
        html = htmls.get()

        soup = BeautifulSoup(html, 'lxml')
        # xpath = '//h2[@class="news_entry"]/a/text()'
        # selector = 'h2.news_entry > a'
        titles = soup.find_all('h2', class_='news_entry')
        for title in titles:
            a = title.a
            if a:
                href = BASE_URL + a.get('href')
                txt = a.txt
                val = href, txt
                outputs.put(val)
                print(val)

# start_urls(1, 1)
# crawler()
# parse()

event = Event()

# 线程池
executor = ThreadPoolExecutor(10)

executor.submit(starts_url, 1, 1)
for i in range(5):
    executor.submit(crawler, event)
for i in range(5):
    executor.submit(parse, event)

import requests
from lxml import etree
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
from threading import Event
import json

BASE_URL = 'https://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
headers = {
  'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
}

urls = Queue() # # 待爬取url队列 # 未来准备用 MQ中间件 替换
htmls = Queue()
outputs = Queue()

# 创建博客园的新闻urls，每页30条新闻
def start_urls(start=1, stop=1, step=1):
    for i in range(start, stop+1, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        urls.put(url)

    print('链接创建完成')

def crawler(e:Event):
    while not e.is_set():
        url = urls.get() # blocked
        response = requests.get(url, headers=headers)
        with response:
            if response.status_code == 200:
                html = response.text
                htmls.put(html)

def parse(e:Event):
    while not e.is_set():
        html = htmls.get() # 阻塞

        # xpath = '//h2[@class="news_entry"]/a/text()'
        # selector = 'h2.news_entry > a'
        # find_all
        doc = BeautifulSoup(html, 'lxml')
        h2s = doc.find_all('h2', class_='news_entry')
        for i, h in enumerate(h2s, 1):
            a = h.a # h.find('a')
            print(i, a.get('href'), a.text) # => 数据可以存储到mongodb中
            val = {
                'href': BASE_URL + a.get('href'),
                'title': a.text
            }
            # Queue中python自己序列化 pickle,但是第三方未必如此
            outputs.put(val)


def persist(e:Event, path:str):
    with open(path, 'a+', encoding='utf-8') as f:
        while not e.is_set():
            val = outputs.get() # dict
            f.write(json.dumps(val))
            f.flush()

# start_urls(1, 1)
# crawler()
# parse()

event = Event()
# 线程池
executor = ThreadPoolExecutor(10)

executor.submit(start_urls,1, 1)
executor.submit(persist, event, 'd:/tmp/titles.txt')
for i in range(5):
    executor.submit(crawler, event)
for i in range(4):
    executor.submit(parse, event)

#生产者代码
import pika
import time

params = pika.URLParameters(
    'amqp://jasper:[email protected]:5672/test'
)

# direct路由，交换机 queue 由消费者随机生成名称绑定rk，生产者exchange 生成时写rk
connection = pika.BlockingConnection(params)
channel = connection.channel()
ex_name = 'news'
ex_type = 'direct'
rks = ('urls', 'htmls', 'outputs')

#生成交换机
channel.exchange_declare(
    exchange=ex_name,
    exchange_type=ex_type)
q_name = rks[0]

#生成队列，并绑定
result = channel.queue_declare(q_name, exclusive=False)
channel.queue_bind(q_name, ex_name)

for i in range(50):
    msg = '{}-data-{:2}'.format(q_name, i)
    channel.basic_publish(exchange=ex_name, # 不指定，默认缺省 test/default
                          routing_key=q_name, # q的名字
                          body=msg)
    print(msg)
    time.sleep(0.5)



connection.close()

#消费者代码
import pika
import time

params = pika.URLParameters(
    'amqp://jasper:[email protected]:5672/test'
)

# direct路由，交换机 queue 由消费者随机生成名称绑定rk，生产者exchange 生成时写rk
connection = pika.BlockingConnection(params)
channel = connection.channel()
ex_name = 'news'
ex_type = 'direct'
rks = ('urls', 'htmls', 'outputs')

#生成交换机
channel.exchange_declare(
    exchange=ex_name,
    exchange_type=ex_type)
q_name = rks[0]

#生成队列，并绑定
result = channel.queue_declare(q_name, exclusive=False)
channel.queue_bind(q_name, ex_name)

def callback(ch, method, properties, body):
    print(f"{method.consumer_tag} -- {body}")

#消费队列
# method, properties, body = channel.basic_get(q_name,True)
channel.basic_consume(
    q_name, on_message_callback=callback, auto_ack=True
)

print(" [*] Waiting for messages. To exit press CTRL+C")
channel.start_consuming()

#创建包文件
./mq/__init__.py

#生产者代码
import pika
import time

class Base:
    def __init__(self, host, vh, exchange, queue, username, password, port=5672, **kwargs):
        params = pika.URLParameters(
            'amqp://{}:{}@{}:{}/{}'.format(
                username,password,host,port,vh
            )
        )

        # direct路由，交换机 queue 由消费者随机生成名称绑定rk，生产者exchange 生成时写rk
        self.connection = pika.BlockingConnection(params)
        self.channel = self.connection.channel()
        self.exchange = exchange
        self.queue = queue
        # ex_name = 'news'
        ex_type = 'direct'
        rks = ('urls', 'htmls', 'outputs')

        #生成交换机
        self.channel.exchange_declare(
            exchange=exchange,
            exchange_type=ex_type)
        # q_name = rks[0]

        #生成队列，并绑定
        result = self.channel.queue_declare(queue, exclusive=False)
        self.channel.queue_bind(queue, exchange)

    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.channel.close()
        self.connection.close()

class Producer(Base):
    def produce(self, message:str):
        self.channel.basic_publish(
            exchange=self.exchange,  # 不指定，默认缺省 test/default
            routing_key=self.queue,  # q的名字
            body=message)

import time
class Consummer(Base):
    def consume(self):
        method, properties, body = self.channel.basic_get(self.queue,True) # 非阻塞
        if body:
            return body
        else:
            time.sleep(1)
            # print('~~~~')

if __name__ == '__main__':
    rks = ('urls', 'htmls', 'outputs')

    #host, vh, exchange, queue, username, password, port=5672,
    with Producer('192.168.226.130', 'test', 'news', 'urls', 'jasper', 123456) as p:
        for i in range(5):
            msg = '{}-data-{:2}'.format(p.queue, i)
            p.produce(msg)

    # request --> response
    with Consummer('192.168.226.130', 'test', 'news', 'urls', 'jasper', 123456) as c:
        with Producer('192.168.226.130', 'test', 'news', 'htmls', 'jasper', 123456) as p:
            for i in range(6):
                url = c.consume()
                if url:
                    # 爬取
                    html = 'aa'
                    p.produce(html)
                print(url, '+++')

    # parse
    with Consummer('192.168.226.130', 'test', 'news', 'htmls', 'jasper', 123456) as c:
        with Producer('192.168.226.130', 'test', 'news', 'outputs', 'jasper', 123456) as p:
            for i in range(6):
                html = c.consume()
                if html:
                    # 解析
                    value = {}
                    p.produce(value)

    with Consummer('192.168.226.130', 'test', 'news', 'outputs', 'jasper', 123456) as c:
        for i in range(6):
            value = c.consume()
            if value:
                # 入库
                pass

import requests
from bs4 import BeautifulSoup
from bs4.element import Tag
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
from threading import Event
import simplejson
from messagequeue import Producer, Consumer


BASE_URL = 'http://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
# https://news.cnblogs.com/n/page/2/ 列表页
# https://news.cnblogs.com/n/628919/ 详情页

headers = {
    'User-agent': "Mozilla/5.0 (Windows; U; Windows NT 6.1; zh-CN) AppleWebKit/537.36 (KHTML,like Gecko)"
                  " Version / 5.0.1Safari / 537.36"
}

# 异步，队列，以后换成第三方队列
# urls = Queue()# 待爬取队列
# htmls = Queue() # 待分析队列
# outputs = Queue() # 待输出队列

# 创建博客园的新闻urls，每页30条新闻
def starts_url(start, stop, step=1):
    p = Producer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    for i in range(start, stop + 1, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        print(url)
        # urls.put(url)  # 加入队列
        p.produce(url)
    print('任务链接创建完毕')


# 爬取页面
def crawler(e:Event):
    p = Producer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    c = Consumer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    while not e.wait(1):
        # url = urls.get()  # 阻塞，拿一条
        url = c.consume()
        if url:
            with requests.get(url, headers=headers) as response:
                if response.status_code == 200:
                    html = response.text
                    #htmls.put(html)
                    p.produce(html)


# 解析页面
def parse(e:Event):
    # 解析
    p = Producer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    c = Consumer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    while not e.wait(1):
        # html = htmls.get()
        html = c.consume()

        if html:
            soup = BeautifulSoup(html, 'lxml')
            # h2.news_entry > a
            # //h2[@new_entry=""]/a
            titles = soup.select('h2.news_entry > a')
            for title in titles:
                val = simplejson.dumps({
                    'title': title.text,
                    'url': BASE_URL + title.get('href','')
                })
                # outputs.put(val)
                p.produce(val)
                # print(val)

# 持久化
def persist(path, e: Event):
    # 以后持久化到数据库当中去
    c = Consumer('192.168.1.5', 5672, 'lqx', 'lqx', 'test', 'news', 'urls')
    with open(path, 'a+', encoding='utf-8') as f:
        while not e.wait(1):
            # val = outputs.get()
            data = c.consume()
            if data:
                val = simplejson.loads(data)
                f.write("{}\x01{}\n".format(val['url'], val['title']))
                f.flush()

event = Event()

# 线程池
executor = ThreadPoolExecutor(10)

executor.submit(starts_url, 1, 2)
executor.submit(persist, 'd:/news.txt', event)

for i in range(5):
    executor.submit(crawler, event)
for i in range(4):
    executor.submit(parse, event)

import requests
from lxml import etree
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
from threading import Event
import json
from mq import Producer, Consummer
import pymongo

BASE_URL = 'https://news.cnblogs.com'
NEWS_PAGE = '/n/page/'
headers = {
  'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36',
}

# urls = Queue() # # 待爬取url队列 # 未来准备用 MQ中间件 替换
# htmls = Queue()
# outputs = Queue()

# 创建博客园的新闻urls，每页30条新闻
def start_urls(start=1, stop=1, step=1):
    p = Producer('192.168.226.130', 'test', 'news', 'urls', 'jasper', 123456)
    for i in range(start, stop+1, step):
        url = "{}{}{}/".format(BASE_URL, NEWS_PAGE, i)
        # urls.put(url)
        p.produce(url)

    print('链接创建完成')

# 爬取页面
def crawler(e:Event):
    c = Consummer('192.168.226.130', 'test', 'news', 'urls', 'jasper', 123456)
    p = Producer('192.168.226.130', 'test', 'news', 'htmls', 'jasper', 123456)
    while not e.wait(1):
        # url = urls.get() # blocked
        url = c.consume()
        if url:
            response = requests.get(url, headers=headers)
            with response:
                if response.status_code == 200:
                    html = response.text
                    # htmls.put(html)
                    p.produce(html)

# 解析页面
def parse(e:Event):
    c = Consummer('192.168.226.130', 'test', 'news', 'htmls', 'jasper', 123456)
    p = Producer('192.168.226.130', 'test', 'news', 'outputs', 'jasper', 123456)
    while not e.is_set():
        # html = htmls.get() # 阻塞
        html = c.consume()
        if html:
            # xpath = '//h2[@class="news_entry"]/a/text()'
            # selector = 'h2.news_entry > a'
            # find_all
            doc = BeautifulSoup(html, 'lxml')
            h2s = doc.find_all('h2', class_='news_entry')
            for i, h in enumerate(h2s, 1):
                a = h.a # h.find('a')
                print(i, a.get('href'), a.text) # => 数据可以存储到mongodb中
                val = {
                    'id': i,
                    'href': BASE_URL + a.get('href'),
                    'title': a.text
                }
                # Queue中python自己序列化 pickle,但是第三方未必如此
                # outputs.put(val)
                p.produce(json.dumps(val))

# 持久化
def persist(e:Event, path:str):
    c = Consummer('192.168.226.130', 'test', 'news', 'outputs', 'jasper', 123456)
    client = pymongo.MongoClient(
        'mongodb://192.168.226.130:27017'
    )
    db = client.news  # 库
    href = db.href  # 表，集合，n条 文档
    while not e.wait(1):
        value = c.consume()
        if value:
            try:
                href.insert_one(json.loads(value))
            except Exception as ex:
                print(ex)
                print(value)

    # with open(path, 'ab') as f:
    #     while not e.is_set():
    #         # val = outputs.get() # dict
    #         value = c.consume()
    #         if value:
    #             # print(value)
    #             f.write(value + b',\n')
    #             f.flush()

event = Event()
# 线程池
executor = ThreadPoolExecutor(10)

executor.submit(start_urls,1, 2)
executor.submit(persist, event, 'd:/tmp/titles.txt')
for i in range(5):
    executor.submit(crawler, event)
for i in range(4):
    executor.submit(parse, event)

element = driver.find_element(By.XPATH, 'xpath_expression')
element.click()

element = driver.find_element(By.XPATH, 'xpath_expression')
element.send_keys("要输入的字符串")

driver = webdriver.Chrome()
# 做几项网络操作，最好是可以前往新页面的操作
driver.back()
driver.forward()

driver.execute_script("window.scrollTo(0,document.body.scrollHeight)")
#JS代码，通过修改scrollHeight的值可以调整滚动的位置，如将scrollHeight修改为0则表示滚动至页面顶部。

page_source = driver.page_source
print(page_source)

element = driver.find_element(By.XPATH, 'xpath_expression')
element_html = element.get_attribute("outerHTML")
print(element_html)
#返回指定元素的HTML代码片段，包括元素本身及其所有子元素

url = driver.current_url
print(url)

element = driver.find_element(By.XPATH, 'xpath_expression')
ActionChains(driver).context_click(element).perform()

element = driver.find_element(By.XPATH, 'xpath_expression')
ActionChains(driver).move_to_element(element).perform()

source_element = driver.find_element(By.XPATH, 'xpath_expression')
target_element = driver.find_element(By.XPATH, 'xpath_expression')
ActionChains(driver).drag_and_drop(source_element, target_element).perform()

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

element = WebDriverWait(driver, 10).until(
    EC.presence_of_element_located((By.XPATH, 'xpath_expression'))
)
代码说明：
WebDriverWait类用于设置最长等待时间，并提供了多种条件来等待特定条件达成。
EC.presence_of_element_located()是一个预定义的条件，用于等待元素出现在页面中。

driver.implicitly_wait(10)

driver.set_page_load_timeout(10)
如果页面在指定时间内未完全加载完成，将抛出TimeoutException异常。

is_page_loaded = driver.execute_script("return document.readyState") == "complete"

# 使用close()方法关闭当前窗口：
driver.close()
# 使用quit()方法退出整个浏览器会话：
driver.quit()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from urllib import parse

import datetime
import random
import time

# 创建ChromeOptions对象，配置Chrome Headless选项
options = webdriver.ChromeOptions()
options.add_argument('--headless') # 设置Chrome为Headless模式
# options.add_argument("--disable-gpu")  # 禁用GPU加速

driver = webdriver.Chrome(options=options)
# 设置窗口大小
driver.set_window_size(1280, 1924)

# 保存图片
def savepic():
    base_path = 'd:/tmp/'
    filename = '{}{:%Y%m%d%H%M%S}{:03}.png'.format(
        base_path,datetime.datetime.now(),random.randint(1, 100)
    )
    driver.save_screenshot(filename)

# 打开网页GET方法，模拟浏览器地址栏输入网址
url = 'https://www.baidu.com/s?' + parse.urlencode({
    'wd': 'python'
})
driver.get(url)
print(url) # https://www.baidu.com/s?wd=python

time.sleep(1)
savepic()

MAXRETRIES = 6 # 最大尝试次数
for i in range(MAXRETRIES):
    time.sleep(4)
    try:
        ele = driver.find_element(By.ID,  'content_left') # 如果查询结果来了，就会有这个id的标签
        if not ele.is_displayed(): # 等待数据显示出来
            print('display none')
            continue
        print('displayed')
        savepic()
        break
    except Exception as e:
        print(e)

driver.quit()

# 获取bing查询数据
from selenium import webdriver # 核心对象
import datetime
import random
import time
from urllib import parse

# 指定PhantomJS的执行文件路径
driver = webdriver.PhantomJS('d:/python/phantomjs/bin/phantomjs.exe')
# 设置窗口大小
driver.set_window_size(1280, 1024)

# 打开网页GET方法，模拟浏览器地址栏输入网址
url = 'http://www.bing.com/search?' + parse.urlencode({
    'q': 'python'
})
driver.get(url)

# 保存图片
def savepic():
    base_dir = 'd:/'
    filename = '{}{:%Y%m%d%H%M%S}{:03}.png'.format(
        base_dir,
        datetime.datetime.now(),
        random.randint(1, 100)
    )

    driver.save_screenshot(filename)

#time.sleep(10) # 等等截图就能看到内容
savepic()

MAXRETRIES = 5 # 最大重试次数
for i in range(MAXRETRIES):  # 循环测试
    time.sleep(1)
    try:
        ele = driver.find_element_by_id('b_results')  # 如果查询结果来了，就会有这个id的标签
        if not ele.is_displayed():  # 等待数据显示出来
            print('diplay none')
            continue
        print('ok')
        savepic()
        break
    except Exception as e:
        print(e)

driver.quit()

# oschina软件搜索
<select name='tag1' onchange="submit();">
    <option value='0'>所有分类</option>
    <option value='309' >Web应用开发</option> 
    <option value='331' >手机/移动开发</option> 
    <option value='364' >iOS代码库</option> 
    <option value='12' >程序开发</option> 
    <option value='11' >开发工具</option> 
    <option value='273' >jQuery 插件</option> 
    <option value='256' >建站系统</option> 
    <option value='5' >企业应用</option> 
    <option value='10' >服务器软件</option> 
    <option value='6' >数据库相关</option> 
    <option value='8' >应用工具</option> 
    <option value='18' >插件和扩展</option> 
    <option value='7' >游戏/娱乐</option>
    <option value='14' >管理和监控</option> 
    <option value='9' >其他开源</option>
</select>

from selenium import webdriver # 核心对象
import datetime
import random
from selenium.webdriver.support.ui import Select

# 指定PhantomJS的执行文件路径
driver = webdriver.PhantomJS('d:/python/phantomjs/bin/phantomjs.exe')
# 设置窗口大小
driver.set_window_size(1280, 1024)

# 保存图片
def savepic():
    base_dir = 'd:/'
    filename = '{}{:%Y%m%d%H%M%S}{:03}.png'.format(
        base_dir,
        datetime.datetime.now(),
        random.randint(1, 100)
    )

    driver.save_screenshot(filename)

# 打开网页GET方法，模拟浏览器地址栏输入网址
url = 'https://www.oschina.net/search?q=python&scope=project'
driver.get(url)

ele = driver.find_element_by_name('tag1') # 获取元素
print(ele.tag_name) # 标签名
print(driver.current_url) # 当前url
savepic()

s = Select(ele)
#s.select_by_index(1)
s.select_by_value('309')
print(driver.current_url) # 新页面
savepic()

driver.quit()

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options
from urllib import parse

import datetime
import random
import time

# 创建ChromeOptions对象，配置Chrome Headless选项
options = webdriver.ChromeOptions()
options.add_argument('--headless') # 设置Chrome为Headless模式
# options.add_argument("--disable-gpu")  # 禁用GPU加速

driver = webdriver.Chrome(options=options)
# 设置窗口大小
driver.set_window_size(1920, 1080)

# 保存图片
def savepic():
    base_path = 'd:/tmp/'
    filename = '{}{:%Y%m%d%H%M%S}{:03}.png'.format(
        base_path,datetime.datetime.now(),random.randint(1, 100)
    )
    driver.save_screenshot(filename)

url = 'https://www.oschina.net/'
driver.get(url)
savepic()

# 点击登录页
login_button = driver.find_element(By.XPATH, '//div[@class="user-bar"]/ul/a')
print(login_button.get_attribute('href')) # 获取属性值
login_button.click() # 点击元素
print(driver.current_url) # 当前url
savepic()

# 登录页，切换标签页
tab = driver.find_element(By.XPATH, '//div[@data-osc-tab="password"]')
tab.click()
savepic()

# 用户名和密码
username = driver.find_element(By.XPATH, '//input[@name="username"]')
password = driver.find_element(By.XPATH, '//input[@name="password"]')
username.send_keys('1xxxx') # 输入用户名
password.send_keys('1xxxx')
savepic()

password.send_keys(Keys.ENTER) # 模拟键盘敲击
# 或者合并 password.send_keys('1xxxx', Keys.ENTER)
time.sleep(5) # 等待2秒
print(driver.current_url)
savepic()

# 模拟登录后获得cookies
cookies = driver.get_cookies()
print(type(cookies), cookies)
for c in cookies:
    print(type(c), c)
print('=' * 30)

# 模拟请求
import requests
from requests.cookies import RequestsCookieJar

# cookies = [
#     {'domain': '.oschina.net', 'httpOnly': False, 'name': 'Hm_lpvt_a411c4d1664dd70048ee98afe7b28f0b', 'path': '/',
#      'sameSite': 'Lax', 'secure': False, 'value': '1757401509'},
# {'domain': '.oschina.net', 'expiry': 1788937508, 'httpOnly': True, 'name': 'oscid', 'path': '/', 'sameSite': 'Lax',
#  'secure': False,
#  'value': 'H1%2FKYVEN1WFHSFPDrSA9WFqLGKWIss6348jpkpX6ZA9DObORyc2LnhWySW0ENLj4GqXLr%2FTP6VOvr9IfJ5TwnVBG3CBGsEj5KmkZ5du2cqfS2%2FTtQ4Nrpbczj4EkvwhjZw9Cxw8hQktrhwfqZTVm0w%3D%3D'},
# ]

headers = {
    'referer':'https://www.oschina.net/',
    'User-agent':"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
}

jar = RequestsCookieJar()
for c in cookies:
    jar.set(c.get('name'), c.get('value'))

url = 'https://www.oschina.net/'

# 不带cookie
with requests.get(url, headers=headers) as response:
    with open('d:/tmp/nocookies.html', 'wb') as f:
        f.write(response.content)
# 带cookie
with requests.get(url, headers=headers, cookies=jar) as response:
    with open('d:/tmp/cookies.html', 'wb') as f:
        f.write(response.content)

driver.quit()

# 模拟开源中国登陆
from selenium import webdriver # 核心对象
import datetime
import random
import time
from selenium.webdriver.common.keys import Keys

# 指定PhantomJS的执行文件路径
driver = webdriver.PhantomJS('d:/python/phantomjs/bin/phantomjs.exe')
# 设置窗口大小
driver.set_window_size(1280, 1024)

# 保存图片
def savepic():
    base_dir = 'd:/'
    filename = '{}{:%Y%m%d%H%M%S}{:03}.png'.format(
        base_dir,
        datetime.datetime.now(),
        random.randint(1, 100)
    )

    driver.save_screenshot(filename)

# 打开网页GET方法，模拟浏览器地址栏输入网址
url = 'https://www.oschina.net/home/login'
driver.get(url)
print(driver.current_url) # 当前url
savepic()

# 模拟输入用户名密码
username = driver.find_element_by_id('userMail') # 获取元素
username.send_keys('[email protected]')
password = driver.find_element_by_id('userPassword') # 获取元素
password.send_keys('cici.com18')
savepic()

password.send_keys(Keys.ENTER)
print('-' * 30)

for i in range(5):
    time.sleep(1)
    try:
        # xpath定位数据
        ele = driver.find_element_by_xpath('//div[@title="cici_wayne"]')
        print(ele.tag_name, '!~~~~~~')
        print(ele.get_attribute('data-user-id'))
        savepic()
        break
    except Exception as e:
        print(e)

# 模拟登录后获得cookies
cookies = driver.get_cookies()
print(cookies)

for c in cookies:
    print(type(c))
    print(c)
print('-' * 30)

import requests
from requests.cookies import RequestsCookieJar

jar = RequestsCookieJar()
for c in cookies:
    jar.set(c.get('name'), c.get('value'))

print(jar)

headers = {'User-agent': "Mozilla/5.0 (Windows; U; "
"Windows NT 6.1; zh-CN) AppleWebKit/537.36(KHTML, like Gecko) Version / 5.0.1Safari / 537.36"}

print('========== 不带cookie ==========')
response = requests.get(url, headers=headers)
with response:
    print(10, response.url)  # 这就是登录页
    with open('d:/nocookie.html', 'w', encoding='utf-8') as f:
        f.write(response.text)

print('========== 带cookie ==========')
response = requests.get(url, headers=headers, cookies=jar)
with response:
    print(11, response.url)
    with open('d:/withcookie.html', 'w', encoding='utf-8') as f:
        f.write(response.text)

driver.quit()