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 # 初始化数据库 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[该消息是接受好友自动回复]" 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) # tgMessage.sendTg(message) else: print('openid未关联') message = f'openid: {openid}-未关联,在{current_time}访问网页' print(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) }