提交代码

.gitignore

@@ -0,0 +1,213 @@
+### JetBrains template
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio, WebStorm and Rider
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/usage.statistics.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# Generated files
+.idea/**/contentModel.xml
+
+# Sensitive or high-churn files
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn.  Uncomment if using
+# auto-import.
+# .idea/artifacts
+# .idea/compiler.xml
+# .idea/jarRepositories.xml
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
+# *.iml
+# *.ipr
+
+# CMake
+cmake-build-*/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+
+# Android studio 3.1+ serialized cache file
+.idea/caches/build_file_checksums.ser
+
+### Python template
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+

Dockerfile

@@ -0,0 +1,48 @@
+# 拉取镜像
+FROM centos:centos7.9.2009
+
+# 配置容器时间
+RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && echo 'Asia/Shanghai' >/etc/timezone
+# 添加快捷命令
+RUN echo "alias ll='ls -hall'" >> ~/.bashrc && source ~/.bashrc
+
+# 更新yum源
+RUN curl -o /etc/yum.repos.d/CentOS7-Aliyun.repo http://mirrors.aliyun.com/repo/Centos-7.repo && curl -o /etc/yum.repos.d/epel-7-Aliyun.repo http://mirrors.aliyun.com/repo/epel-7.repo
+RUN yum clean all && yum makecache && yum -y update
+RUN yum install -y kde-l10n-Chinese
+
+# 设置系统编码
+ENV LANG=zh_CN.UTF-8
+# 设置vi编码（防止中文乱码）
+RUN grep -qxF 'set encoding=utf8' /etc/virc || echo 'set encoding=utf8' >> /etc/virc
+
+WORKDIR /opt
+# 安装 python3.8.10 gcc相关配置
+RUN yum --exclude=kernel* update -y && yum groupinstall -y 'Development Tools' && yum install -y gcc openssl-devel bzip2-devel libffi-devel mesa-libGL
+# python3.8.10下载与解压缩
+RUN curl -o python3.8.10.tgz https://mirrors.huaweicloud.com/python/3.8.10/Python-3.8.10.tgz && tar -zxvf python3.8.10.tgz
+
+# 设置Python3虚拟环境路径
+ENV VIRTUAL_ENV=/usr/local/python38
+# 切换路径
+WORKDIR /opt/Python-3.8.10
+# 创建安装目录, 指定安装目录
+RUN mkdir $VIRTUAL_ENV && ./configure --prefix=$VIRTUAL_ENV
+# 编译、安装
+RUN make -j 8 && make altinstall
+# 创建python3的软连接
+RUN rm -rf /usr/bin/python3 /usr/bin/pip3 && ln -s $VIRTUAL_ENV/bin/python3.8 /usr/bin/python3 && ln -s $VIRTUAL_ENV/bin/pip3.8 /usr/bin/pip3
+# 更换pip源&更新pip
+RUN pip3 config set global.index-url https://mirrors.aliyun.com/pypi/simple/ && pip3 install --upgrade pip
+# python3虚拟环境加入系统环境变量
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+
+# 指定应用安装目录
+WORKDIR /app
+# 复制当前目录文件到容器/app目录下
+COPY requirements.txt requirements.txt
+# 安装python项目依赖
+RUN pip3 install -r requirements.txt
+
+# 指定工作目录
+WORKDIR /mnt

README.md

@@ -0,0 +1,21 @@
+# 验证码识别服务
+
+#### 创建docker容器
+```shell
+   docker build -t pycaptcha:latest .
+```
+
+#### 开启服务
+```shell
+   docker-compose up -d
+```
+
+#### 关闭服务并删除运行容器
+```shell
+   docker-compose down
+```
+
+#### 重启服务
+```shell
+   docker-compose restart
+```

build_app.py

@@ -0,0 +1,68 @@
+import os
+
+from fastapi import FastAPI, Request
+from fastapi.openapi.docs import (
+    get_redoc_html,
+    get_swagger_ui_html,
+    get_swagger_ui_oauth2_redirect_html,
+)
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+
+from exception import ApiException
+from services.limiter import register_limiter
+from services.routers import service_router
+
+
+def register_swagger_ui_html(app):
+    app.mount("/static", StaticFiles(directory="static"), name="static")
+
+    @app.get("/docs", include_in_schema=False)
+    async def custom_swagger_ui_html():
+        return get_swagger_ui_html(
+            openapi_url=app.openapi_url,
+            title=app.title + " - Swagger UI",
+            oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
+            swagger_js_url="/static/swagger-ui-bundle.js",
+            swagger_css_url="/static/swagger-ui.css",
+        )
+
+    @app.get(app.swagger_ui_oauth2_redirect_url, include_in_schema=False)
+    async def swagger_ui_redirect():
+        return get_swagger_ui_oauth2_redirect_html()
+
+    @app.get("/redoc", include_in_schema=False)
+    async def redoc_html():
+        return get_redoc_html(
+            openapi_url=app.openapi_url,
+            title=app.title + " - ReDoc",
+            redoc_js_url="/static/redoc.standalone.js",
+        )
+
+
+def init_app():
+    env = os.getenv('FASTAPI_ENV')
+    if env != 'prod':
+        app = FastAPI(title="crawl_server", docs_url=None, redoc_url=None)
+        register_swagger_ui_html(app)
+    else:
+        # 生产环境关闭文档说明
+        app = FastAPI(title="crawl_server", docs_url=None, redoc_url=None)
+
+    app.include_router(service_router, prefix="/v1")
+
+    @app.exception_handler(ApiException)
+    async def api_exception(request: Request, exc: ApiException):
+        return JSONResponse(status_code=200, content=dict(code=exc.code, errMsg=exc.errMsg, r=exc.r))
+
+    @app.exception_handler(Exception)
+    async def unknown_exception(request: Request, exc: Exception):
+        return JSONResponse(status_code=500, content=dict(code=1, errMsg="服务器内部错误,暂无法提供服务", r={}))
+
+    return app
+
+
+def create_app():
+    app = init_app()
+    register_limiter(app)
+    return app

captcha_server.py

@@ -0,0 +1,8 @@
+from build_app import create_app
+
+app = create_app()
+
+
+# if __name__ == '__main__':
+#     import uvicorn
+#     uvicorn.run(app="captcha_server:app", host="0.0.0.0", port=2119, reload=True, debug=True)

chaojiying.py

@@ -0,0 +1,67 @@
+# coding:utf-8
+
+from hashlib import md5
+
+import requests
+
+
+class ChaoJiYIngClient:
+
+    def __init__(self, username, password, soft_id):
+        self.username = username
+        password = password.encode('utf8')
+        self.password = md5(password).hexdigest()
+        self.soft_id = soft_id
+        self.base_params = {
+            'user': self.username,
+            'pass2': self.password,
+            'softid': self.soft_id,
+        }
+        self.headers = {
+            'Connection': 'Keep-Alive',
+            'User-Agent': 'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0)',
+        }
+
+    def postpic(self, im, codetype):
+        """
+        im: 图片字节
+        codetype: 题目类型 参考 http://www.chaojiying.com/price.html
+        """
+        params = {
+            'codetype': codetype,
+        }
+        params.update(self.base_params)
+        files = {'userfile': ('ccc.jpg', im)}
+        r = requests.post('http://upload.chaojiying.net/Upload/Processing.php',
+                          data=params, files=files, headers=self.headers,
+                          timeout=60)
+        return r.json()
+
+    def postpic_base64(self, base64_str, codetype):
+        """
+        im: 图片字节
+        codetype: 题目类型 参考 http://www.chaojiying.com/price.html
+        """
+        params = {
+            'codetype': codetype,
+            'file_base64': base64_str
+        }
+        params.update(self.base_params)
+        r = requests.post('http://upload.chaojiying.net/Upload/Processing.php',
+                          data=params, headers=self.headers, timeout=60)
+        return r.json()
+
+    def report_error(self, im_id):
+        """
+        im_id:报错题目的图片ID
+        """
+        params = {
+            'id': im_id,
+        }
+        params.update(self.base_params)
+        r = requests.post('http://upload.chaojiying.net/Upload/ReportError.php',
+                          data=params, headers=self.headers, timeout=60)
+        return r.json()
+
+
+CJ = ChaoJiYIngClient('ddddjy', 'T95jJcRu57sFAFn', '929622')

docker-compose.yml

@@ -0,0 +1,27 @@
+version: "3" # docker版本对应的docker-compose版本
+services:
+  server:
+    container_name: pycaptcha
+    image: pycaptcha:latest
+    volumes:
+      - /mnt/pycaptcha:/mnt
+    network_mode: "host"
+    restart: always
+    ulimits:
+      core: 0  # 禁止生成core文件
+    tmpfs:
+      - /tmp
+    env_file:
+      - ./web.env
+    privileged: true
+    shm_size: 10GB
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "200k"
+        max-file: "10"
+    deploy:
+      resources:
+        limits:
+          memory: 10G
+    command: 'python3 -m gunicorn -c gunicorn.conf.py captcha_server:app'

exception.py

@@ -0,0 +1,7 @@
+
+class ApiException(Exception):
+
+    def __init__(self, code: int, errMsg: str = "服务器内部错误,暂无法提供服务", r: dict = {}):
+        self.code = code
+        self.errMsg = errMsg
+        self.r = r

glovar.py

@@ -0,0 +1,16 @@
+# 限制器
+lm_counter = 0  # 计数器
+lm_date = None  # 当前日期
+lm_forecast_warning = True  # 上限预告标识
+lm_exceeded_warning = True  # 超限告警标识
+"""
+    参考：https://limits.readthedocs.io/en/latest/quickstart.html
+    Examples:
+        1、10 per hour
+        2、10/hour
+        3、10/hour;100/day;2000 per year
+        4、100/day, 500/7days
+        5、5/minute
+"""
+lm_max_limit = 1000  # 最大限制数量
+lm_value = f'{lm_max_limit}/day'  # 限制数量

gunicorn.conf.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""
+Created on 2023-04-24 
+---------
+@summary:  gunicorn配置
+---------
+@author: Dzr
+"""
+import multiprocessing
+
+# 服务地址
+bind = '0.0.0.0:2119'
+# 代码更改时重新启动工作程序(适用于开发测试)
+reload = False
+# 日志输出级别
+loglevel = 'info'
+# 访问记录到标准输出
+accesslog = '-'
+# 访问记录格式
+access_log_format = '%(h)s  %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"'
+# 工作进程数量
+workers = multiprocessing.cpu_count() * 2 + 1
+# 工作线程数量(当工作模式指定为gevent evenlet等异步类型时，线程变成基于Greentlet的task(伪线程)，这时候threads参数是无效的)
+# threads = multiprocessing.cpu_count() * 2
+# 工作模式
+worker_class = 'uvicorn.workers.UvicornWorker'

libs/tj_arithmetic/charsets.json

@@ -0,0 +1 @@
+{"charset": [" ", "0", "6", "+", "2", "1", "3", "4", "-", "8", "9", "7", "5", "x"], "image": [-1, 64], "word": false, "channel": 1}

requirements.txt

@@ -0,0 +1,11 @@
+ddddocr==1.4.7
+fastapi==0.68.0
+slowapi==0.1.5
+opencv-python==4.6.0.66
+numpy==1.24.1
+requests==2.26.0
+uvicorn==0.15.0
+gunicorn==20.1.0
+aiofiles==23.1.0
+python-multipart==0.0.6
+Pillow==9.5.0

services/defults.py

@@ -0,0 +1,7 @@
+# 用户数据库
+FAKE_USERS_DB = {
+    "jianyu001": {
+        "username": "jianyu001",
+        "hashed_password": "123qwe!A",
+    }
+}

services/images/apis.py

@@ -0,0 +1,197 @@
+import base64
+import io
+import re
+import time
+
+import cv2
+import ddddocr
+import numpy as np
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query
+from fastapi.requests import Request
+from fastapi.security import OAuth2PasswordRequestForm
+
+import glovar
+from chaojiying import CJ
+from services.defults import FAKE_USERS_DB
+from services.limiter import LIMITER, limiter_warring, flush_limiter
+from services.utils import calculate
+
+images_router = APIRouter(prefix="/images")
+
+
+def verify_identity(form_data: OAuth2PasswordRequestForm):
+    user_dict = FAKE_USERS_DB.get(form_data.username)
+    if not user_dict:
+        raise HTTPException(status_code=400, detail="用户名或密码错误")
+
+
+@images_router.post("/verify", summary="ocr")
+async def simple_captcha(file: UploadFile = File(...)):
+    start = time.time()
+    img_bytes = await file.read()
+    ocr = ddddocr.DdddOcr(det=False, ocr=True, show_ad=False)
+    if img_bytes.startswith(b'data:image'):
+        src = img_bytes.decode()
+        result = re.search("data:image/(?P<ext>.*?);base64,(?P<data>.*)", src, re.DOTALL)
+        if result:
+            # ext = result.groupdict().get("ext")
+            img_base64 = result.groupdict().get("data")
+        else:
+            raise Exception("Do not parse!")
+        verify_code = ocr.classification(img_base64)
+    else:
+        verify_code = ocr.classification(img_bytes)
+    spend = "{:.2f}".format(time.time() - start)
+    return {
+        "msg": "success",
+        "code": 0,
+        "r": {
+            'time': float(spend),
+            'code': verify_code
+        }
+    }
+
+
+@images_router.post("/arithmetic", summary="100以内算术")
+async def arithmetic_captcha(file: UploadFile = File(...)):
+    start = time.time()
+    img_bytes = await file.read()
+    onnx_path = 'libs/tj_arithmetic/tj_project_1.0_23_15000_2023-01-14-10-58-23.onnx'
+    charsets_path = 'libs/tj_arithmetic/charsets.json'
+    ocr = ddddocr.DdddOcr(det=False, ocr=False,show_ad=False,
+                          import_onnx_path=onnx_path,
+                          charsets_path=charsets_path)
+    if img_bytes.startswith(b'data:image'):
+        src = img_bytes.decode()
+        result = re.search("data:image/(?P<ext>.*?);base64,(?P<data>.*)", src, re.DOTALL)
+        if result:
+            # ext = result.groupdict().get("ext")
+            img_base64 = result.groupdict().get("data")
+        else:
+            raise Exception("Do not parse!")
+        verify_code = ocr.classification(img_base64)
+    else:
+        verify_code = ocr.classification(img_bytes)
+
+    verify_code = verify_code.replace('x', '*')
+
+    spend = "{:.2f}".format(time.time() - start)
+    return {
+        "msg": "success",
+        "code": 0,
+        "r": {
+            'time': float(spend),
+            'code': calculate(verify_code)
+        }
+    }
+
+
+@images_router.post("/verify_det", summary="点验")
+async def det_captcha(
+    image_content: UploadFile = File(..., description='验证码图片')
+):
+    det = ddddocr.DdddOcr(det=True, show_ad=False)
+    ocr = ddddocr.DdddOcr(ocr=True, show_ad=False)
+    img_bytes = await image_content.read()
+    if img_bytes.startswith(b'data:image'):
+        src = img_bytes.decode()
+        result = re.search("data:image/(?P<ext>.*?);base64,(?P<data>.*)", src, re.DOTALL)
+        if result:
+            img_base64 = result.groupdict().get("data")
+        else:
+            raise Exception("Do not parse!")
+        poses = det.detection(img_base64=img_base64)
+        img_bytes = base64.b64decode(img_base64)
+    else:
+        poses = det.detection(img_bytes=img_bytes)
+    img_byte = io.BytesIO(img_bytes)
+    file_array = np.frombuffer(img_byte.getbuffer(), np.uint8)
+    image = cv2.imdecode(file_array, cv2.IMREAD_COLOR)
+    strxys = {}
+    for box in poses:
+        # 对框内文字进行识别
+        x1, y1, x2, y2 = box
+        part = image[y1:y2, x1:x2]
+        img = cv2.imencode('.jpg', part)[1]
+        result = ocr.classification(img.tobytes())
+        result = re.sub("[a-zA-Z0-9]+", "", result)
+        if len(result) > 1:
+            result = result[0]
+        strxys[result] = [x1, y1, x2, y2]
+    result = {
+        "msg": "success",
+        "code": 0,
+        "r": {
+            "code": strxys,
+            "code_list": poses}
+    }
+    return result
+
+
+@images_router.get("/reset", summary="解除限制")
+async def reset_limiter():
+    LIMITER.reset()
+    flush_limiter()
+    return {
+        "msg": "success",
+        "code": 0,
+        "r": {}
+    }
+
+
+async def cjy_postpic_base64(
+    pic_type: str = Query(..., min_length=4, max_length=4, description='验证码图片类型'),
+    file: UploadFile = File(..., description='验证码图片'),
+):
+    start = time.time()
+    img_bytes = await file.read()
+    base64_str = base64.b64encode(img_bytes)
+    discern_result = CJ.postpic_base64(base64_str, int(pic_type))
+    spend = "{:.2f}".format(time.time() - start)
+    err_no = discern_result['err_no']
+    pic_id = discern_result['pic_id']
+    result = {
+        "msg": discern_result['err_str'],
+        "code": err_no,
+        "r": {
+            'time': float(spend),
+            'pic_str': discern_result['pic_str'],
+            'pic_id': pic_id,
+            'md5': discern_result['md5'],
+        }
+    }
+    return result
+
+
+@images_router.post("/discern", summary="超级鹰")
+@LIMITER.limit(glovar.lm_value)
+async def discern_complex_image(
+    request: Request,
+    pic_type: str,
+    file: UploadFile = File(..., description='验证码图片'),
+    form_data: OAuth2PasswordRequestForm = Depends()
+):
+    verify_identity(form_data)
+    result = await cjy_postpic_base64(pic_type=pic_type, file=file)
+    glovar.lm_counter += 1
+    limiter_warring()
+    return result
+
+
+@images_router.post("/report_err", summary="上传超级鹰识别错误")
+async def cjy_report_error(
+    pic_id: str = Query(
+        ...,
+        description='验证码图片id',
+    ),
+    form_data: OAuth2PasswordRequestForm = Depends()
+):
+    user_dict = FAKE_USERS_DB.get(form_data.username)
+    if not user_dict:
+        raise HTTPException(status_code=400, detail="用户名或密码错误")
+    report_error = CJ.report_error(pic_id)
+    return {
+        "msg": report_error['err_str'],
+        "code": report_error['err_no'],
+        "r": {}
+    }

services/limiter.py

@@ -0,0 +1,45 @@
+from fastapi import FastAPI
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
+
+import glovar
+from services.robot import send_msg
+from services.utils import current_date
+
+LIMITER = Limiter(key_func=get_remote_address)
+MAX_LIMIT = glovar.lm_max_limit
+
+
+def limit_exceeded_handler(*args, **kwargs):
+    if glovar.lm_exceeded_warning:
+        msg = "今日接口调用次数已达上限！\n 继续使用请点击"
+        send_msg('超级鹰', MAX_LIMIT, MAX_LIMIT, msg, allow_reset=True)
+        glovar.lm_exceeded_warning = False
+    return _rate_limit_exceeded_handler(*args, **kwargs)
+
+
+def register_limiter(app: FastAPI):
+    app.state.limiter = LIMITER
+    app.add_exception_handler(RateLimitExceeded, limit_exceeded_handler)
+
+
+def flush_limiter():
+    glovar.lm_counter = 0
+    glovar.lm_forecast_warning = True
+    glovar.lm_exceeded_warning = True
+
+
+def limiter_warring():
+    platform = '超级鹰'
+    curr_date = current_date()
+    if curr_date != glovar.lm_date:
+        if glovar.lm_date is not None:
+            flush_limiter()
+            glovar.lm_counter += 1
+        glovar.lm_date = curr_date
+    val = '{:.2f}'.format(glovar.lm_counter / MAX_LIMIT * 100)
+    if float(val) > 80 and glovar.lm_forecast_warning:
+        msg = f'使用次数已超过{val}%'
+        send_msg(platform, MAX_LIMIT, glovar.lm_counter, msg)
+        glovar.lm_forecast_warning = False

services/robot.py

@@ -0,0 +1,60 @@
+import requests
+
+import platform
+
+# 企业微信联系人
+MOBILE_LIST = [
+    "15639297172",  # 张金坤
+]
+
+if platform.system() not in ['Darwin', 'Windows']:
+    PLATFORM_RUL = "http://123.57.163.80:2119"
+else:
+    PLATFORM_RUL = "http://127.0.0.1:2119"
+
+
+def post_data(platform, limit, usage, msg, allow_reset=False):
+    """
+
+    :param platform:平台名称
+    :param limit:限制次数
+    :param usage:使用次数
+    :param msg: 通知内容
+    :param allow_reset: 解除限制
+    :return:
+    """
+    label = "<font color=\"warning\">{}</font>".format(msg)
+    platform = ">平台名称: <font color=\"comment\">{}</font>".format(platform)
+    c0 = ">调用上限: <font color=\"comment\">{}次</font>".format(limit)
+    c1 = ">调用次数: <font color=\"comment\">{}次</font>".format(usage)
+    msg = "{}\n {}\n > {}\n > {}\n".format(label, platform, c0, c1)
+    if allow_reset:
+        c2 = f"[解除限制]({PLATFORM_RUL}/v1/images/reset)"
+        msg = "{}[{}]\n > {}\n > {}\n {}".format(label, c2, platform, c0, c1)
+
+    content = msg
+    markdown = {"msgtype": "markdown", "markdown": {"content": content}}
+    text = {
+        "msgtype": "text",
+        "text": {
+            "content": " 请以下同事注意",
+            "mentioned_mobile_list": MOBILE_LIST
+        }
+    }
+    body = {'markdown': markdown, 'text': text}
+    return body
+
+
+def robot_webhook(data):
+    url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
+    headers = {"Content-Type": "application/json"}
+    params = {"key": "683a19fe-c72d-464f-acbe-489656f06b05"}
+    response = requests.post(url, headers=headers, params=params, json=data)
+    return response.json()
+
+
+def send_msg(platform, limit, usage, msg, **kwargs):
+    contents = post_data(platform, limit, usage, msg, **kwargs)
+    for _, content in contents.items():
+        response = robot_webhook(content)
+        print('企业微信机器人 >>> ', response)

services/routers.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+from services.images.apis import images_router
+
+
+service_router = APIRouter()
+service_router.include_router(images_router, tags=["验证码识别"])

services/utils.py

@@ -0,0 +1,14 @@
+import ast
+import datetime
+
+
+def current_date(fmt="%Y-%m-%d"):
+    return datetime.datetime.now().strftime(fmt)
+
+
+def literal_eval(node_or_string):
+    return ast.literal_eval(node_or_string)
+
+
+def calculate(string):
+    return eval(compile(string, "<string>", "eval"))

web.env

@@ -0,0 +1,4 @@
+# 生产环境
+FASTAPI_ENV=prod
+# 测试环境
+# FASTAPI_ENV=develop