update

tmp/wechat/push_user.py

@@ -1,7 +1,23 @@
 import json
+import random
 
 import requests
 
+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_QiYeVX(corp_id, agent_id, secret, _message, userid_list=None):  # 默认发送给自己
     if userid_list is None:
@@ -13,10 +29,13 @@ def send_message_QiYeVX(corp_id, agent_id, secret, _message, userid_list=None):
 
     json_dict = {
         "touser": userid_str,
-        "msgtype": "text",
+        "msgtype": "textcard",
         "agentid": agent_id,
-        "text": {
-            "content": _message
+        "textcard": {
+            "title": "念念不忘，必有回响",
+            "description": get_des(),
+            "url": "https://memory.tianyunperfect.cn/",
+            "btntxt": "更多"
         },
         "safe": 0,
         "enable_id_trans": 0,

tmp/wechat/push_user_app.py

@@ -0,0 +1,147 @@
+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):
+    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": "念念不忘，必有回响",
+            "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)])}
+
+
+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