rate-sys-template/Context/common.py

# 全局使用的服务放到这里初始化，作为项目配置输入的入口
from configparser import ConfigParser
from logging import getLogger
from sqlalchemy.orm import declarative_base
from ThirdPartyApi.Tianyancha.Apis import TianyanchaApi
from Utils.AuthUtils import AuthUtil, TokenDataModel
from Utils.MongoUtils import MongoConnect
from Utils.PhoneMsgUtils import PhoneMsgSend
from Utils.RedisUtils import RedisPool
from Utils.SqlAlchemyUtils import SqlalchemyConnect
from Utils.VerifyCodeUtils import EmailCodeVerify, ImageCaptchaVerify, PhoneVerifyCode
from pydantic import BaseModel
from typing import Optional

# 权限验证工具类

# 日志
uvicorn_log = getLogger('uvicorn')

# 配置数据
conf = ConfigParser()
conf.read("Config/common.ini", encoding='utf-8-sig')


class CompanyUserTokenDataModel(BaseModel):
    id: Optional[str]
    company_name: Optional[str]
    email: Optional[str]
    phone: Optional[str]


class InstitutionUserTokenDataModel(BaseModel):
    id: Optional[str]
    institution_name: Optional[str]
    email: Optional[str]
    phone: Optional[str]


class AdminUserTokenDataModel(BaseModel):
    id: Optional[str]
    email: Optional[str]
    phone: Optional[str]


auth_util = AuthUtil[TokenDataModel](secret_key=conf['auth']['jwt_key'], tokenDataModel=TokenDataModel)
# 管理端用户权限
admin_user_auth_util = AuthUtil[AdminUserTokenDataModel](secret_key=conf['auth']['admin_user_jwt_key'],
                                                         tokenDataModel=AdminUserTokenDataModel)
# 企业用户权限
company_user_auth_util = AuthUtil[CompanyUserTokenDataModel](secret_key=conf['auth']['company_user_jwt_key'],
                                                             tokenDataModel=CompanyUserTokenDataModel)
# 协会用户权限
institution_user_auth_util = AuthUtil[InstitutionUserTokenDataModel](
    secret_key=conf['auth']['institution_user_jwt_key'],
    tokenDataModel=InstitutionUserTokenDataModel)

##############################################################


# redis连接池
redis_pool = RedisPool(host=conf['redis']['host'], port=int(conf['redis']['port']))
redis_pool.connect()

# 共用mysql数据库
common_db_base = declarative_base()
common_db = SqlalchemyConnect(common_db_base, host=conf['mysql']['host'],
                              user=conf['mysql']['user'],
                              password=conf['mysql']['password'],
                              db=conf['mysql']['db'])

# mongo db
mg_db = MongoConnect(conf["mongo"]["host"], int(conf["mongo"]["port"]),
                     conf["mongo"]["db"],
                     conf["mongo"].get('user'),
                     conf["mongo"].get("password"))
# 邮箱验证码工具类
email_code_verify = EmailCodeVerify(redis_pool, conf['email']['api'], conf['email']['sender'])

# 图片验证码工具类
image_captcha_verify = ImageCaptchaVerify(redis_pool)

# 短信发送工具
phone_msg_send = PhoneMsgSend(conf['hw_msg']['url'],
                              conf['hw_msg']['app_key'],
                              conf['hw_msg']['app_secret'],
                              conf['hw_msg']['sender'],
                              conf['hw_msg']['template_id'],
                              conf['hw_msg']['signature']
                              )

# 短信验证码工具
phone_verify_code = PhoneVerifyCode(redis_pool, phone_msg_send)

# 天眼查接口
tianyancha_api = TianyanchaApi(conf['tianyancha']['token'], mongo_connect=mg_db)

# 全局配置