初始化投流业务框架

README.md

@@ -1,3 +1,19 @@
 # TouliuServer
 
-投流数据文章相关服务开发
+投流数据文章相关服务开发
+
+### 1. LogServer
+###### code: 日志码
+    rangs 1001 --> 9999
+    具体细节待设计
+###### env: 日志环境
+    prod: 正式环境
+    dev: 测试环境
+###### message: 日志信息
+    日志基础信息，类似于print
+###### route: 接口信息
+    用于划分某个具体的接口所执行的功能
+###### trace_id： 追踪id
+    touliu + "_" + uuid + "_" + timestamp
+###### data: 日志附加信息
+    一般存储传参，返回结果等json信息

applications/__init__.py

@@ -0,0 +1,5 @@
+"""
+@author: luojunhui
+"""
+from .aliyunLog import log
+from .asyncDB import AsyncMySQLClient

applications/aliyunLog.py

@@ -0,0 +1,57 @@
+"""
+@author: luojunhui
+"""
+import time
+import json
+from aliyun.log import LogClient, PutLogsRequest, LogItem
+
+
+def log(
+        code,
+        env="prod",
+        trace_id=None,
+        route=None,
+        message=None,
+        data=None
+):
+    """
+    :return:
+    """
+    if data is None:
+        data = {}
+    accessKeyId = "LTAIP6x1l3DXfSxm"
+    accessKey = "KbTaM9ars4OX3PMS6Xm7rtxGr1FLon"
+    project = "changwen-alg"
+    log_store = "touliu_server"
+    endpoint = "cn-hangzhou.log.aliyuncs.com"
+
+    # 创建 LogClient 实例
+    client = LogClient(endpoint, accessKeyId, accessKey)
+    log_group = []
+    log_item = LogItem()
+    contents = [
+        (f"env", str(env)),
+        (f"code", str(code)),
+        (f"route", str(route)),
+        (f"message", str(message)),
+        (f"data", json.dumps(data, ensure_ascii=False) if data else ""),
+        (f"trace_id", str(trace_id)),
+        ("timestamp", str(int(time.time()))),
+    ]
+
+    log_item.set_contents(contents)
+    log_group.append(log_item)
+    # 写入日志
+    request = PutLogsRequest(
+        project=project,
+        logstore=log_store,
+        topic="",
+        source="",
+        logitems=log_group,
+        compress=False,
+    )
+    try:
+        client.put_logs(request)
+    except Exception as e:
+        print("日志失败")
+        print(e)

applications/asyncDB.py

@@ -0,0 +1,93 @@
+"""
+@author: luojunhui
+"""
+import aiomysql
+
+from applications import log
+from config import env
+
+
+class AsyncMySQLClient(object):
+    """
+    Async MySQL
+    """
+
+    def __init__(self, app):
+        self.app = app
+
+    async def initPool(self):
+        """
+        初始化连接
+        :return:
+        """
+        if env == "prod":
+            self.app.mysql_pool = await aiomysql.create_pool(
+                host='rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com',
+                port=3306,
+                user='crawler',
+                password='crawler123456@',
+                db='piaoquan-crawler',
+                charset='utf8mb4',
+                maxsize=100,
+                connect_timeout=120,
+            )
+            log(
+                code="1001",
+                env=env,
+                message="数据库初始化成功"
+            )
+        elif env == "dev":
+            self.app.mysql_pool = await aiomysql.create_pool(
+                host='rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com',
+                port=3306,
+                user='crawler',
+                password='crawler123456@',
+                db='piaoquan-crawler',
+                charset='utf8mb4',
+                maxsize=100,
+                connect_timeout=120,
+            )
+            log(
+                code="1001",
+                env=env,
+                message="数据库初始化成功"
+            )
+        else:
+            log(
+                code="1001",
+                env=env,
+                message="数据库初始化失败，环境校验失败"
+            )
+            return None
+
+    async def closePool(self):
+        """
+        关闭 mysql 连接
+        :return:
+        """
+        self.app.mysql_pool.close()
+        await self.app.mysql_pool.wait_closed()
+
+    async def asyncSelect(self, sql):
+        """
+        select method
+        :param sql:
+        :return:
+        """
+        async with self.app.mysql_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute(sql)
+                result = await cursor.fetchall()
+                return result
+
+    async def asyncInsert(self, sql, params):
+        """
+        insert and update method
+        :param params:
+        :param sql:
+        :return:
+        """
+        async with self.app.mysql_pool.acquire() as coon:
+            async with coon.cursor() as cursor:
+                await cursor.execute(sql, params)
+                await coon.commit()

config/__init__.py

@@ -0,0 +1,6 @@
+"""
+@author: luojunhui
+config
+"""
+# 环境
+env = "prod"

requirements.txt

@@ -0,0 +1,11 @@
+quart
+hypercorn
+aiomysql~=0.2.0
+pymysql
+aiohttp
+requests
+numpy
+pandas
+aliyun-log-python-sdk
+aliyun-python-sdk-core
+aliyun-python-sdk-kms

routes/__init__.py

@@ -0,0 +1,35 @@
+"""
+@author: luojunhui
+投流--路由
+"""
+from quart import Blueprint, jsonify, request
+
+from config import env
+
+TL_blueprint = Blueprint("TouLiu", __name__)
+
+
+def Routes(db_client):
+    """
+    路由代码
+    :param db_client: 异步db连接池
+    :return:
+    """
+
+    @TL_blueprint.route("/hello")
+    def helloWorld():
+        """
+        :return: Hello World
+        """
+        return jsonify({"message": "Hello World!"})
+
+    @TL_blueprint.route("/generateInfo", methods=['POST'])
+    async def generateInfo():
+        """
+        生成一些需要用到到信息
+        :return:
+        """
+        return jsonify({"message": "this function is developing"})
+
+    return TL_blueprint
+

touliu.toml

@@ -0,0 +1,6 @@
+reload = true
+bind = "0.0.0.0:8813"
+workers = 4
+keep_alive_timeout = 120  # 保持连接的最大秒数，根据需要调整
+graceful_timeout = 30    # 重启或停止之前等待当前工作完成的时间
+loglevel = "debug"  # 日志级别

touliu_app.py

@@ -0,0 +1,37 @@
+"""
+@author: luojunhui
+"""
+from quart import Quart
+from routes import Routes
+from applications import AsyncMySQLClient
+
+# init app
+app = Quart(__name__)
+
+asyncMysql = AsyncMySQLClient(app)
+
+appRoutes = Routes(asyncMysql)
+
+app.register_blueprint(appRoutes)
+
+
+@app.before_serving
+async def initDB():
+    """
+    init db pool before server is up
+    :return:
+    """
+    await asyncMysql.initPool()
+
+
+@app.after_serving
+async def closeDB():
+    """
+    close db pool after server is down
+    :return:
+    """
+    await asyncMysql.closePool()
+
+
+if __name__ == '__main__':
+    app.run(debug=True)