python-base--/project/push_user_app.py

import json
import random
import threading
import time

import requests
import uvicorn
from fastapi import FastAPI
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import JSONResponse

app = FastAPI(title="push_user_app ", description="通用系统 ", version="v 0.0.0")

# 添加 session 中间键，使项目中可以使用session
app.add_middleware(SessionMiddleware, secret_key='123456hh112312h')
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*", ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    """接口服务异常捕获; text/plain"""
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """请求校验异常捕获; application/json """
    json_response = JSONResponse({'message': '【 {} 】接口触发了 RequestValidationError 错误.'
                                 .format(request.scope.get("endpoint").__name__, exc.errors()), 'status_code': 999,
                                  "error_message": exc.errors(), "queryParasm": exc.body})
    return json_response


@app.exception_handler(Exception)
async def validation_exception_handler(request, exc):
    """请求校验异常捕获; application/json """
    return JSONResponse({'message': "服务器内部错误", 'status_code': 500})


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    """接口响应中间键; 当前只支持 http 请求"""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["API-Process-Time"] = f"{process_time} seconds"
    return response


# http://localhost:8000/query?uid=1
@app.get("/query")
def query(uid):
    msg = f'uid为{uid}'
    return {'success': True, 'msg': msg}


token = None


def delete_token():
    global token
    time.sleep(60 * 60)
    token = None


notes = [
    """我们曾如此渴望命运的波澜，到最后才发现，人生最曼妙的风景，竟是内心的淡定与从容；
    我们曾如此期盼外界的认可，到最后才知道，世界是自己的，与他人毫无关系。""",
    """做一个浪漫的人，对世界保持永远的好奇，但别让他轻易改变你。""",
    """我们终其一生都是为了获得选择的权利和拒绝的底气。""",
    """无聊是非常有必要的，一个人在空白时间所做的事，决定了这个人和他人根本的不同。""",
    """当爱支配一切时，权力就不存在了；
    当权力主宰一切时，爱就消失了。两者互为对方的影子。""",
    """没有一个人是住在客观的世界里， 我们都居住在一个各自赋予其意义的主观的世界。""",
    """每天安静地坐十五分钟，倾听你的气息，感觉它，感觉你自己，并且试着什么都不想。"""
]


def get_des():
    li = random.choice(notes).split("\n")
    li.append("点击进行复习。")
    res = ''
    for line in li:
        res += "<div>" + line.strip() + "</div>"
    return res


def send_message(corp_id, agent_id, secret, userid_list=None, title=None):
    global token

    if userid_list is None:
        userid_list = []

    userid_str = "|".join(userid_list)
    if token is None:
        print("获取一次 token")
        response = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corp_id}&corpsecret={secret}")
        data = json.loads(response.text)
        access_token = data['access_token']
        token = access_token
        threading.Thread(target=delete_token).start()
    else:
        access_token = token

    json_dict = {
        "touser": userid_str,
        "msgtype": "textcard",
        "agentid": agent_id,
        "textcard": {
            "title": title,
            "description": get_des(),
            "url": "https://memory.tianyunperfect.cn/",
            "btntxt": "更多"
        },
        "safe": 0,
        "enable_id_trans": 0,
        "enable_duplicate_check": 0,
        "duplicate_check_interval": 1800
    }
    json_str = json.dumps(json_dict)
    response_send = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}",
                                  data=json_str)
    return json.loads(response_send.text)['errmsg'] == 'ok'


@app.get("/push_user")
def push_user(user_id):
    return {
        'success': send_message("ww582505d35b886cde", "1000002", "yDCwvVOHSvb_43Y3e17mZi4E7hEZ2Z3UDyDpuKxzPsQ",
                                [str(user_id)], "念念不忘，必有回响")}


@app.get("/push_msg")
def push_msg(user_id, title):
    return {
        'success': send_message("ww582505d35b886cde", "1000002", "yDCwvVOHSvb_43Y3e17mZi4E7hEZ2Z3UDyDpuKxzPsQ",
                                [str(user_id)], title)}


if __name__ == '__main__':
    uvicorn.run(app='push_user_app:app', host="0.0.0.0", port=8087, reload=True, debug=True)

# uvicorn push_user_app:app --reload --host 0.0.0.0 --port 9100