174 lines
5.8 KiB
Python
174 lines
5.8 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
|
||
|
||
# 初始化数据库
|
||
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"}
|
||
|
||
|
||
@app.get("/hello")
|
||
def hello():
|
||
response = f"你好,\n..........................\n一名开发者\n副业:跨境电商\n#公众号:异度世界\n..........................\n回复关键词,了解更多:\n1. 微信机器人\n2. 数字游民\n3. 加群\n..........................\n[该消息是接受好友自动回复]"
|
||
return PlainTextResponse(content=response)
|
||
|
||
|
||
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("用户已存在")
|
||
return "用户已存在"
|
||
db.add(user)
|
||
db.commit()
|
||
except IntegrityError:
|
||
db.rollback()
|
||
print("用户已存在")
|
||
return "用户已存在"
|
||
|
||
|
||
@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,
|
||
db: Session = Depends(get_db),
|
||
background_tasks: BackgroundTasks = None):
|
||
|
||
user_data = schemas.UserCreate(nickname=nickname,
|
||
remark=remark,
|
||
wechat_number=wechat_number,
|
||
wxid=wxid,
|
||
gender=gender,
|
||
friendaddtime=friendaddtime,
|
||
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[该消息是接受好友自动回复]"
|
||
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")
|
||
return 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 crud.update_user(db=db, user=user, user_update=user_update)
|
||
|
||
|
||
@app.get("/generate_qr")
|
||
def generate_qr(wxid: Optional[str] = Query(None), db: Session = Depends(get_db)):
|
||
# 选择网络图片
|
||
# return FileResponse('images/111.png')
|
||
|
||
|
||
|
||
url = "https://web.hackrobot.cn"
|
||
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 FileResponse(combined_image_path) |