基于go-cqhttp與Flask搭建定制機(jī)器人項(xiàng)目實(shí)戰(zhàn)示例

更新時(shí)間：2022年11月23日 09:59:03 作者：????????????

這篇文章主要為大家介紹了基于go-cqhttp與Flask搭建定制機(jī)器人項(xiàng)目實(shí)戰(zhàn)示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

許久之前用 Mirai 搭建了 QQ 機(jī)器人，不過(guò)因?yàn)樵品?wù)器到期了，QQ 機(jī)器人被迫下線，現(xiàn)如今，可能是意猶未盡，今天就基于 go-cqhttp 與 Flask 搭建定制機(jī)器人；

安裝

官方安裝教程；

go-cqhttp 下載地址；

1、前往 go-cqhttp 的下載地址，選擇自己需要的版本，下載的話，一般下載這兩個(gè)中的其中一個(gè)：

具體看自己的系統(tǒng)配置，可以通過(guò) arch 指令進(jìn)行查看：

博主這里是 x86_64，所以選擇下載第一個(gè)，也就是 go-cqhttp_linux_amd64.tar.gz；

x86_64，x64，AMD64 可看做是同一個(gè)東西；

x86 是 intel 開(kāi)發(fā)的一種32位指令集；
x86_64 是一種64位的指令集，x86_64 是 x86 指令的超集，在 x86 上可以運(yùn)行的程序，在 x86_64 上也可以運(yùn)行，x86_64 是 AMD 發(fā)明的，也叫 AMD64；

現(xiàn)在用的 intel/amd 的桌面級(jí) CPU 基本上都是 x86_64，與之相對(duì)的 arm 等都不是 x86_64；

2、創(chuàng)建并打開(kāi) go-cqhttp 文件夾：

cd /opt && mkdir go-cqhttp

3、然后把自己下載好的壓縮包上傳到此處，或者直接通過(guò) wget 指令將壓縮包下載到此處；

wget url

4、解壓壓縮包：

tar -zxvf go-cqhttp_linux_amd64.tar.gz

5、賦予 go-cqhttp 權(quán)限：

chmod 777 go-cqhttp

6、第一次運(yùn)行，生成配置文件：

./go-cqhttp

7、修改配置文件，具體參考官方配置文檔；

# go-cqhttp 默認(rèn)配置文件
account: # 賬號(hào)相關(guān)
  uin: 1233456 # QQ賬號(hào)
  password: '' # 密碼為空時(shí)使用掃碼登錄
  encrypt: false  # 是否開(kāi)啟密碼加密
  status: 0      # 在線狀態(tài) 請(qǐng)參考 https://docs.go-cqhttp.org/guide/config.html#在線狀態(tài)
  relogin: # 重連設(shè)置
    delay: 3   # 首次重連延遲, 單位秒
    interval: 3   # 重連間隔
    max-times: 0  # 最大重連次數(shù), 0為無(wú)限制
  # 是否使用服務(wù)器下發(fā)的新地址進(jìn)行重連
  # 注意, 此設(shè)置可能導(dǎo)致在海外服務(wù)器上連接情況更差
  use-sso-address: true
heartbeat:
  # 心跳頻率, 單位秒
  # -1 為關(guān)閉心跳
  interval: 5
message:
  # 上報(bào)數(shù)據(jù)類型
  # 可選: string,array
  post-format: string
  # 是否忽略無(wú)效的CQ碼, 如果為假將原樣發(fā)送
  ignore-invalid-cqcode: false
  # 是否強(qiáng)制分片發(fā)送消息
  # 分片發(fā)送將會(huì)帶來(lái)更快的速度
  # 但是兼容性會(huì)有些問(wèn)題
  force-fragment: false
  # 是否將url分片發(fā)送
  fix-url: false
  # 下載圖片等請(qǐng)求網(wǎng)絡(luò)代理
  proxy-rewrite: ''
  # 是否上報(bào)自身消息
  report-self-message: false
  # 移除服務(wù)端的Reply附帶的At
  remove-reply-at: false
  # 為Reply附加更多信息
  extra-reply-data: false
  # 跳過(guò) Mime 掃描, 忽略錯(cuò)誤數(shù)據(jù)
  skip-mime-scan: false
output:
  # 日志等級(jí) trace,debug,info,warn,error
  log-level: warn
  # 日志時(shí)效 單位天. 超過(guò)這個(gè)時(shí)間之前的日志將會(huì)被自動(dòng)刪除. 設(shè)置為 0 表示永久保留.
  log-aging: 15
  # 是否在每次啟動(dòng)時(shí)強(qiáng)制創(chuàng)建全新的文件儲(chǔ)存日志. 為 false 的情況下將會(huì)在上次啟動(dòng)時(shí)創(chuàng)建的日志文件續(xù)寫(xiě)
  log-force-new: true
  # 是否啟用 DEBUG
  debug: false # 開(kāi)啟調(diào)試模式
# 默認(rèn)中間件錨點(diǎn)
default-middlewares: &default
  # 訪問(wèn)密鑰, 強(qiáng)烈推薦在公網(wǎng)的服務(wù)器設(shè)置
  access-token: ''
  # 事件過(guò)濾器文件目錄
  filter: ''
  # API限速設(shè)置
  # 該設(shè)置為全局生效
  # 原 cqhttp 雖然啟用了 rate_limit 后綴, 但是基本沒(méi)插件適配
  # 目前該限速設(shè)置為令牌桶算法, 請(qǐng)參考:
  # https://baike.baidu.com/item/%E4%BB%A4%E7%89%8C%E6%A1%B6%E7%AE%97%E6%B3%95/6597000?fr=aladdin
  rate-limit:
    enabled: false # 是否啟用限速
    frequency: 1  # 令牌回復(fù)頻率, 單位秒
    bucket: 1     # 令牌桶大小
# 連接服務(wù)列表
servers:
  # HTTP 通信設(shè)置
  - http:
      # 服務(wù)端監(jiān)聽(tīng)地址
      host: 127.0.0.1
      # 服務(wù)端監(jiān)聽(tīng)端口
      port: 5700
      # 反向HTTP超時(shí)時(shí)間, 單位秒
      # 最小值為5，小于5將會(huì)忽略本項(xiàng)設(shè)置
      timeout: 5
      middlewares:
        <<: *default # 引用默認(rèn)中間件
      # 反向HTTP POST地址列表
      post:
      #- url: '' # 地址
      #  secret: ''           # 密鑰
      #- url: 127.0.0.1:5701 # 地址
      #  secret: ''          # 密鑰
  # 正向WS設(shè)置
  - ws:
      # 正向WS服務(wù)器監(jiān)聽(tīng)地址
      host: 127.0.0.1
      # 正向WS服務(wù)器監(jiān)聽(tīng)端口
      port: 6700
      middlewares:
        <<: *default # 引用默認(rèn)中間件
  - ws-reverse:
      # 反向WS Universal 地址
      # 注意 設(shè)置了此項(xiàng)地址后下面兩項(xiàng)將會(huì)被忽略
      universal: ws://your_websocket_universal.server
      # 反向WS API 地址
      api: ws://your_websocket_api.server
      # 反向WS Event 地址
      event: ws://your_websocket_event.server
      # 重連間隔 單位毫秒
      reconnect-interval: 3000
      middlewares:
        <<: *default # 引用默認(rèn)中間件
  # pprof 性能分析服務(wù)器, 一般情況下不需要啟用.
  # 如果遇到性能問(wèn)題請(qǐng)上傳報(bào)告給開(kāi)發(fā)者處理
  # 注意: pprof服務(wù)不支持中間件、不支持鑒權(quán). 請(qǐng)不要開(kāi)放到公網(wǎng)
  - pprof:
      # pprof服務(wù)器監(jiān)聽(tīng)地址
      host: 127.0.0.1
      # pprof服務(wù)器監(jiān)聽(tīng)端口
      port: 7700
  # LambdaServer 配置
  - lambda:
      type: scf # 可用 scf,aws (aws未經(jīng)過(guò)測(cè)試)
      middlewares:
        <<: *default # 引用默認(rèn)中間件
  # 可添加更多
  #- ws-reverse:
  #- ws:
  #- http:
database: # 數(shù)據(jù)庫(kù)相關(guān)設(shè)置
  leveldb:
    # 是否啟用內(nèi)置leveldb數(shù)據(jù)庫(kù)
    # 啟用將會(huì)增加10-20MB的內(nèi)存占用和一定的磁盤(pán)空間
    # 關(guān)閉將無(wú)法使用 撤回 回復(fù) get_msg 等上下文相關(guān)功能
    enable: true

8、再次運(yùn)行 ## go-cqhttp 掃碼綁定機(jī)器人；

9、另起會(huì)話 screen -S sid10t-bot，將其在后臺(tái)掛起；

發(fā)送消息

access_token = ''
url = f"http://127.0.0.1:port/send_private_msg?access_token={access_token}"
msg = "Hello World! --sid10t."
data = {
    "user_id": QQ Number,
    "message": msg
}
requests.post(url, data=data, timeout=5)

接收消息

這里的話不用別人的插件，自己寫(xiě) Flask 監(jiān)聽(tīng)端口，別問(wèn)為什么不用其他的框架，比如 SpringBoot 之類的，麻煩，每次更新都要重新編譯打包；

先寫(xiě)個(gè) app.py 構(gòu)建基本的 Flask 框架，同時(shí)監(jiān)聽(tīng)自己的 port：

from flask import Flask, request
from api import keyword
app = Flask(__name__)
@app.route('/', methods=["POST", "GET"])
def get_data():
    if request.get_json().get('message_type') == 'private':	
        uid = request.get_json().get('sender').get('user_id')
        msg = request.get_json().get('raw_message')
        keyword(msg, uid)
    if request.get_json().get('message_type')=='group':
        gid = request.get_json().get('group_id')
        uid = request.get_json().get('sender').get('user_id')
        msg = request.get_json().get('raw_message') 
        keyword(msg, uid, gid)
    return "OK"
if __name__ =="__main__":
    app.run(debug=True, port=port)

然后再寫(xiě)一個(gè) api.py 處理收集到的信息：

def keyword(msg, uid, gid=None):
    person_send_hello(uid)
def send_private_msg(uid):
    url = f"{BASEURL}/send_private_msg?access_token={ACCESS_TOKEN}"
    data = {
        "user_id": uid,
        "message": "Hello World! --sid10t."
    }
    try:
        requests.post(url, data=data, timeout=5)
    except:
        pass