278 lines
10 KiB
Python
278 lines
10 KiB
Python
from sqlite3 import IntegrityError
|
||
from fastapi import BackgroundTasks, FastAPI, HTTPException, Depends, Query
|
||
from fastapi.responses import JSONResponse, PlainTextResponse
|
||
from sqlalchemy.orm import Session
|
||
from typing import List, Optional, Any
|
||
|
||
from app import models, schemas, crud
|
||
from app.database import SessionLocal, engine
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from PIL import Image
|
||
import qrcode
|
||
import os
|
||
from fastapi.responses import FileResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
# from app import tgMessage # type: ignore #发送tg消息
|
||
from datetime import datetime
|
||
import requests
|
||
|
||
# 初始化数据库
|
||
models.Base.metadata.create_all(bind=engine)
|
||
|
||
app = FastAPI()
|
||
|
||
# 挂载 images 文件夹
|
||
app.mount("/images", StaticFiles(directory="images"), name="images")
|
||
|
||
# 允许所有来源跨域访问（不推荐用于生产环境）
|
||
origins = ["*"]
|
||
|
||
# 设置CORS（跨域资源共享）策略
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"], # 允许所有来源
|
||
allow_credentials=True,
|
||
allow_methods=["*"], # 允许所有方法
|
||
allow_headers=["*"], # 允许所有请求头
|
||
)
|
||
|
||
|
||
# 依赖注入
|
||
def get_db():
|
||
db = SessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# @app.get("/")
|
||
# def root():
|
||
# return {"message": "Hello World"}
|
||
|
||
def sengTgmesage(message):
|
||
url = "http://192.168.31.38:8521/send_message"
|
||
data = {"message":message}
|
||
headers = {"Content-Type": "application/json"}
|
||
response = requests.post(url, json=data, headers=headers)
|
||
try:
|
||
response_data = response.json()
|
||
print(response_data)
|
||
except requests.exceptions.JSONDecodeError:
|
||
print("Response content is not in JSON format")
|
||
print("Response content:", response.text)
|
||
|
||
|
||
@app.get("/hello")
|
||
def hello():
|
||
# response = f"你好,\n..........................\n一名开发者\n副业:跨境电商\n#公众号:异度世界\n..........................\n回复关键词,了解更多:\n1. 微信机器人\n2. 数字游民\n3. 加群\n..........................\n[该消息是接受好友自动回复]"
|
||
response = f"你好,\n..........................\n一名开发者\n副业:跨境电商\n#公众号:异度世界\n..........................\n回复关键词,了解更多:\n1. 数字游民\n..........................\n[该消息是接受好友自动回复]"
|
||
|
||
return PlainTextResponse(content=response)
|
||
|
||
|
||
@app.get("/tgMessage")
|
||
def tgMessagefun(openid: Optional[str] = None, db: Session = Depends(get_db)):
|
||
current_time = datetime.now()
|
||
print("Current Date and Time:", current_time)
|
||
# 发送tg消息通知 wxid_4413224132412 || o7LFAwR32hWGq9XOpd7ZxK1wZxq8
|
||
|
||
# openid = "o7LFAwR32hWGq9XOpd7ZxK1wZxq8"
|
||
print('openid', openid)
|
||
if openid:
|
||
# 查询用户信息:昵称
|
||
data = read_user(openid, db=db)
|
||
# print('data==>', data)
|
||
userInfo = data['userInfo']
|
||
# print('userInfo==>', userInfo)
|
||
if userInfo:
|
||
# print('userInfo==>', userInfo)
|
||
# message = f"{userInfo['nickname']}/{userInfo['gender']}在{current_time}进行了访问"
|
||
message = f"{userInfo.nickname}/{userInfo.gender}在{current_time}进行了访问"
|
||
print(message)
|
||
sengTgmesage(message)
|
||
|
||
# tgMessage.sendTg(message)
|
||
else:
|
||
print('openid未关联')
|
||
message = f'openid: {openid}-未关联,在{current_time}访问网页'
|
||
print(message)
|
||
sengTgmesage(message)
|
||
|
||
# tgMessage.sendTg(message)
|
||
|
||
|
||
def create_user_task(db: Session, user: schemas.UserCreate):
|
||
try:
|
||
existing_user = db.query(
|
||
models.User).filter(models.User.wxid == user.wxid).first()
|
||
if existing_user:
|
||
# 可以选择更新用户信息或返回错误
|
||
print("用户已存在create_user_task")
|
||
return "用户已存在"
|
||
else:
|
||
user = models.User(nickname=user.nickname,
|
||
remark=user.remark,
|
||
wechat_number=user.wechat_number,
|
||
wxid=user.wxid,
|
||
gender=user.gender,
|
||
friendaddtime=user.friendaddtime,
|
||
openid=user.openid)
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
|
||
except IntegrityError:
|
||
db.rollback()
|
||
print("IntegrityError")
|
||
return "IntegrityError"
|
||
|
||
|
||
# 生成二维码图片
|
||
def generateqr(url="http://nomad.hackrobot.cn",
|
||
wxid=None,
|
||
group_image_path="images/group.jpg"):
|
||
|
||
# url = "http://nomad.hackrobot.cn"
|
||
# url="http://192.168.31.219:10086"
|
||
if wxid:
|
||
url += f"?wxid={wxid}"
|
||
|
||
# 创建二维码
|
||
qr = qrcode.QRCode(version=1, box_size=5, border=2) # 调整二维码大小
|
||
qr.add_data(url)
|
||
qr.make(fit=True)
|
||
img = qr.make_image(fill='black', back_color='white')
|
||
|
||
# 保存二维码图像
|
||
qr_image_path = f"images/{wxid}-qr_code.png"
|
||
img.save(qr_image_path)
|
||
|
||
# 加载原图
|
||
# group_image_path = "images/group.jpg"
|
||
|
||
# 在原图上添加二维码（右下角）
|
||
group_image = Image.open(group_image_path)
|
||
qr_image = Image.open(qr_image_path)
|
||
|
||
# 将二维码放置在右下角
|
||
qr_image = qr_image.resize((200, 200)) # 调整二维码大小
|
||
qr_position = (group_image.width - qr_image.width - 50,
|
||
group_image.height - qr_image.height - 50) # 距离20像素
|
||
# 在原图上添加二维码
|
||
group_image.paste(qr_image, qr_position)
|
||
|
||
# 保存合成的图像
|
||
combined_image_path = f"images/combined_{wxid}.jpg"
|
||
group_image.save(combined_image_path)
|
||
# 压缩图片
|
||
group_image.save(combined_image_path, "JPEG",
|
||
quality=10) # 85 是压缩质量，可根据需要调整
|
||
|
||
# 删除二维码文件
|
||
os.remove(qr_image_path)
|
||
print(combined_image_path)
|
||
return combined_image_path
|
||
# return FileResponse(combined_image_path)
|
||
|
||
|
||
@app.get("/add")
|
||
def add_user(nickname: Optional[Any] = None,
|
||
remark: Optional[Any] = None,
|
||
wechat_number: Optional[Any] = None,
|
||
wxid: Optional[Any] = None,
|
||
gender: Optional[Any] = None,
|
||
friendaddtime: Optional[Any] = None,
|
||
openid: Optional[Any] = None,
|
||
keyword: Optional[Any] = None,
|
||
db: Session = Depends(get_db),
|
||
background_tasks: BackgroundTasks = None):
|
||
response = None
|
||
user_data = schemas.UserCreate(
|
||
nickname=nickname,
|
||
remark=remark,
|
||
wechat_number=wechat_number,
|
||
wxid=wxid,
|
||
gender=gender,
|
||
# friendaddtime=friendaddtime,
|
||
friendaddtime=str(datetime.now()),
|
||
openid=openid)
|
||
if background_tasks:
|
||
background_tasks.add_task(create_user_task, db, user_data)
|
||
# response = f"{nickname}你好,\n..........................\n一名开发者\n副业:跨境电商\n#公众号:异度世界\n..........................\n回复关键词,了解更多:\n1. 微信机器人\n2. 数字游民\n3. 加群\n..........................\n[该消息是接受好友自动回复]"
|
||
response = f"{nickname}你好,\n..........................\n一名开发者\n副业:跨境电商\n#公众号:异度世界\n..........................\n回复关键词,了解更多:\n1. 数字游民\n..........................\n[该消息是接受好友自动回复]"
|
||
|
||
# 关键词回复
|
||
if keyword == '加群' or keyword =="数字游民" or keyword =="3" or keyword =="1":
|
||
print("keyword==加群")
|
||
combined_image_path = generateqr(url="http://nomad.hackrobot.cn",
|
||
wxid=wxid,
|
||
group_image_path="images/nomadqr.jpg")
|
||
return FileResponse(combined_image_path)
|
||
elif keyword == '手册':
|
||
print("keyword==手册")
|
||
response = f"手册正在编写中"
|
||
combined_image_path = generateqr(url="http://nomad.hackrobot.cn",
|
||
wxid=wxid,
|
||
group_image_path="images/bookqr.jpg")
|
||
return FileResponse(combined_image_path)
|
||
elif keyword == '活动':
|
||
print("keyword==活动")
|
||
response = f"活动正在筹备中"
|
||
combined_image_path = generateqr(url="http://nomad.hackrobot.cn",
|
||
wxid=wxid,
|
||
group_image_path="images/meetupqr.jpg")
|
||
return FileResponse(combined_image_path)
|
||
elif keyword == 'qrcode':
|
||
print("keyword==qrcode")
|
||
combined_image_path = generateqr(url="http://nomad.hackrobot.cn",
|
||
wxid='qrcode',
|
||
group_image_path="images/nomadqr.jpg")
|
||
return FileResponse(combined_image_path)
|
||
else:
|
||
print('其他keyword')
|
||
return PlainTextResponse(content=response)
|
||
|
||
|
||
@app.get("/users/")
|
||
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
|
||
return crud.get_users(db=db, skip=skip, limit=limit)
|
||
|
||
|
||
@app.get("/user/{identifier}")
|
||
def read_user(identifier: str, db: Session = Depends(get_db)):
|
||
user = crud.get_user_by_openid_or_wxid(db, identifier=identifier)
|
||
if user is None:
|
||
# raise HTTPException(status_code=404, detail="User not found")
|
||
print('查询用户不存在')
|
||
return {"code": '400', "userInfo": None}
|
||
# return user
|
||
return {"code": '200', "userInfo": user}
|
||
|
||
|
||
@app.delete("/user/{user_id}")
|
||
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
||
user = crud.get_user(db=db, user_id=user_id)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return crud.delete_user(db=db, user=user)
|
||
|
||
|
||
@app.put("/user/{identifier}")
|
||
def update_user(identifier: str,
|
||
user_update: schemas.UserUpdate,
|
||
db: Session = Depends(get_db)):
|
||
user = crud.get_user_by_openid_or_wxid(db, identifier=identifier)
|
||
if user is None:
|
||
# raise HTTPException(status_code=404, detail="User not found")
|
||
# return user
|
||
return {"code": '400', "userInfo": None}
|
||
|
||
# return crud.update_user(db=db, user=user, user_update=user_update)
|
||
return {
|
||
"code": '200',
|
||
"userInfo": crud.update_user(db=db, user=user, user_update=user_update)
|
||
}
|
||
|
||
|