v6ole
7455d7e426
130 issues auto-fixed (import ordering, UP045/UP006 type annotations), 33 issues manually fixed (E712/E501/E402/E722 + N818 rename + per-file wechat ignore for N8xx naming conventions). All 33 tests pass.
69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
from fastapi import APIRouter, Request, Response
|
|
|
|
from app.wechat.handler import WeChatMessageHandler
|
|
|
|
router = APIRouter()
|
|
|
|
_handler: WeChatMessageHandler | None = None
|
|
|
|
|
|
def get_handler() -> WeChatMessageHandler:
|
|
global _handler
|
|
if _handler is None:
|
|
_handler = WeChatMessageHandler()
|
|
return _handler
|
|
|
|
|
|
@router.get("/wechat/callback")
|
|
async def wechat_verify(request: Request):
|
|
handler = get_handler()
|
|
params = request.query_params
|
|
echostr = handler.verify_url(
|
|
params.get("msg_signature", ""),
|
|
params.get("timestamp", ""),
|
|
params.get("nonce", ""),
|
|
params.get("echostr", ""),
|
|
)
|
|
if echostr:
|
|
return Response(content=echostr, media_type="text/plain")
|
|
return Response(content="verification failed", status_code=403)
|
|
|
|
|
|
@router.post("/wechat/callback")
|
|
async def wechat_callback(request: Request):
|
|
handler = get_handler()
|
|
params = request.query_params
|
|
post_data = await request.body()
|
|
post_text = post_data.decode("utf-8")
|
|
|
|
xml_tree = handler.decrypt_message(
|
|
post_text,
|
|
params.get("msg_signature", ""),
|
|
params.get("timestamp", ""),
|
|
params.get("nonce", ""),
|
|
)
|
|
if xml_tree is None:
|
|
return Response(content="decrypt failed", status_code=403)
|
|
|
|
msg_type = xml_tree.find("MsgType")
|
|
msg_type = msg_type.text if msg_type is not None else "unknown"
|
|
|
|
if msg_type == "event":
|
|
event = xml_tree.find("Event")
|
|
event_key = xml_tree.find("EventKey")
|
|
from_user = xml_tree.find("FromUserName")
|
|
handler.handle_event(
|
|
event.text if event is not None else "",
|
|
event_key.text if event_key is not None else None,
|
|
from_user.text if from_user is not None else "",
|
|
)
|
|
elif msg_type == "text":
|
|
content = xml_tree.find("Content")
|
|
from_user = xml_tree.find("FromUserName")
|
|
handler.handle_text(
|
|
content.text if content is not None else "",
|
|
from_user.text if from_user is not None else "",
|
|
)
|
|
|
|
return Response(content="success")
|