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 # from tentcentSMS import sendSms import random from tencentcloud.common import credential from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException from tencentcloud.sms.v20210111 import sms_client, models as smsmodels from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile # 初始化数据库 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"} # phoneNum="+8613243889782" def sendSms(phoneNum,verification_code=None): try: cred = credential.Credential("AKIDOaT6SZQ0JL7cqnTryoOhkW1FBqELAizz", "kDPKyMTJs99sKVP1NREkzxKQFo66pIUP") httpProfile = HttpProfile() httpProfile.reqMethod = "POST" # post请求(默认为post请求) httpProfile.reqTimeout = 30 # 请求超时时间，单位为秒(默认60秒) httpProfile.endpoint = "sms.tencentcloudapi.com" # 指定接入地域域名(默认就近接入) clientProfile = ClientProfile() clientProfile.signMethod = "TC3-HMAC-SHA256" # 指定签名算法 clientProfile.language = "en-US" clientProfile.httpProfile = httpProfile client = sms_client.SmsClient(cred, "ap-beijing", clientProfile) req = smsmodels.SendSmsRequest() req.SmsSdkAppId = "1400808237" req.SignName = "异度世界公众号" req.TemplateId = "1795415" #验证码为：{1}, 若非本人操作，请忽略。 req.TemplateParamSet = [verification_code] newphoneNum = '+86' + phoneNum req.PhoneNumberSet = [newphoneNum] req.SessionContext = "" req.ExtendCode = "" req.SenderId = "" resp = client.SendSms(req) print(resp.to_json_string(indent=2)) except TencentCloudSDKException as err: print(err) 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() # 格式化日期和时间，不包含空格 formatted_now = current_time.strftime("%Y%m%d%H%M%S") print("Current Date and Time:", formatted_now) # 发送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}在{formatted_now}进行了访问" print(message) sengTgmesage(message) # tgMessage.sendTg(message) else: print('openid未关联') message = f'openid: {openid}-未关联,在{formatted_now}访问网页' 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) current_time = datetime.now() # 格式化日期和时间，不包含空格 formatted_now = current_time.strftime("%Y%m%d%H%M%S") # 保存合成的图像 # combined_image_path = f"images/combined_{wxid}_{datetime.now()}.jpg" combined_image_path = f"images/combined_{wxid}_{formatted_now}.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().strftime("%Y%m%d%H%M%S")), 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) } # 获取手机号=>生成验证码 @app.post("/get_verification_code/") def get_verification_code(wxid: str, phone_number: str, db: Session = Depends(get_db)): verification_code = '{:04d}'.format(random.randint(0, 9999)) user = db.query(models.User).filter(models.User.wxid == wxid).first() if not user: raise HTTPException(status_code=404, detail="User not found") user.phone_number = phone_number user.verification_code = verification_code db.commit() sendSms(phone_number) return { "message": "Verification code sent", "verification_code": verification_code } # 验证码校验 @app.post("/verify_code/") def verify_code(phone_number: str, code: str, db: Session = Depends(get_db)): user = db.query( models.User).filter(models.User.phone_number == phone_number).first() if not user: raise HTTPException(status_code=404, detail="User not found") if user.verification_code == code: return {"message": "Verification successful"} else: raise HTTPException(status_code=400, detail="Invalid verification code")